消息中间件Rabbit(三)-Simple简单模式

左手的ㄟ右手 2023-10-06 21:08 77阅读 0赞

## 一、simple简单模式 ##

![058cb523a9b64444b9fb14aa121071d1.png][]

这里的“P”是生产者，“C”是消费者。中间的框是一个队列-RabbitMQ，代表使用者保留的消息缓冲区。

## 二、生产者代码实现 ##

2.1 创建一个工程，pom依赖

<dependencies>
            
            <dependency>
                <groupId>com.rabbitmq</groupId>
                <artifactId>amqp-client</artifactId>
                <version>5.14.0</version>
            </dependency>
            
            <dependency>
                <groupId>commons-io</groupId>
                <artifactId>commons-io</artifactId>
                <version>2.6</version>
            </dependency>
        </dependencies>

2.2 公共类，用来创建资源，关闭资源

public class RabbitMqUtils {
        //获取链接对象，本处都是设置的默认值
        public static Connection getConnection() throws IOException, TimeoutException {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("127.0.0.1");
            factory.setPort(5672);
            factory.setUsername("guest");
            factory.setPassword("guest");
            factory.setVirtualHost("/");
            Connection connection=factory.newConnection();
            return connection;
        }
        //释放资源,需要注意的是，虽然将通道关闭掉是最佳实践，但并不是必须的操作。因为无论何种情况，通道都会在底层的连接关闭时自动关闭掉。
        public static void closeConnectionAndChanel(Connection connection,Channel channel) throws IOException, TimeoutException {
            if(connection != null)
            {
                connection.close();
            }
            if(channel != null)
            {
                channel.close();
            }
        }
    }

2.2 生产者代码

public class Producer {
    
        private final static  String QUEUE_NAME="hello";
        public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
            //创建一个链接工厂
            Connection connection= RabbitMqUtils.getConnection();
            Connection connection=factory.newConnection();
            Channel channel = connection.createChannel();
    
            /** queueDeclare声明一个队列，不存在则创建
             * 参数：
             * 1.队列名称
             * 2.队列里的消息是否持久化，默认消息存储在内存中
             * 3.该队列是否只供一个消费者进行消费，是否共享
             * 4.是否自动删除 最后一个消费者断开连接以后，该队列是否自动删除
             * 5.其他参数
             */
            channel.queueDeclare(QUEUE_NAME,false,false,false,null);
    
            String message="Hello world";
            /**发送一个消息
             * 1.发送到哪个交换机
             * 2.路由的key是哪个
             * 3.其他的参数信息
             * 4.消息体
             */
            channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
            System.out.println("发送消息完毕");
            //释放资源
            RabbitMqUtils.closeConnectionAndChanel(connection,channel);
        }
    }

2.3 查看消息队列

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA6Iqx5p6c5rC06I6y_size_20_color_FFFFFF_t_70_g_se_x_16][]

## 三、 消费者的代码实现 ##

3.1 代码实现

public class Concumer {
        private final static  String QUEUE_NAME="hello";
        public static void main(String[] args) throws IOException, TimeoutException{
            Connection connection=RabbitMqUtils.getConnection();
            Channel channel = connection.createChannel();
            System.out.println("等待接收消息");
    
            //推送的消息如何进行消费的接口回调
            DeliverCallback deliverCallback = (consumerTag,delivery)->{
                    String message = new String(delivery.getBody());
                    System.out.println("消费的消息："+message);
                };
    
                CancelCallback cancelCallback = (consumerTag)->{
                  System.out.println("消息消费被中断");
                };
    
                /**消费者消费消息
                 * 1.消费哪个队列
                 * 2.消费成功后是否自动应答
                 * 3.消费者成功消费的回调函数
                 * 4.消费者取消消费的回调函数
                 *
                 */
                channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);
    
        }
    }

注意：消费端的connection和channel是没有释放的，会一直监听队列中的消息。如果释放的话会报错。

3.2 查看消息队列

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA6Iqx5p6c5rC06I6y_size_20_color_FFFFFF_t_70_g_se_x_16 1][]

可以看到消息已经被消费

[058cb523a9b64444b9fb14aa121071d1.png]: https://img-blog.csdnimg.cn/058cb523a9b64444b9fb14aa121071d1.png
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA6Iqx5p6c5rC06I6y_size_20_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/cff3e1c30f4442df97d1c0f3ed852e52.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA6Iqx5p6c5rC06I6y,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA6Iqx5p6c5rC06I6y_size_20_color_FFFFFF_t_70_g_se_x_16 1]: https://img-blog.csdnimg.cn/f5be370fa738428aa4703b4cf0a3f2e8.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA6Iqx5p6c5rC06I6y,size_20,color_FFFFFF,t_70,g_se,x_16