RabbitMQ快速入门

Posted 2020-12-24 shuchen007

• 用处：
  

将一些无需即时返回且耗时的操作提取出来。进行了异步处理，而这种异步处理的方式大大的节省了server的请求响应时间，从而提高了系统的吞吐量。

如登录系统调用短信系统和安全系统，可以通过消息中间件来调用短信和安全系统。

 

队列服务, 会有三个概念： 发消息者、队列、收消息者，

RabbitMQ 在这个基本概念之上, 多做了一层抽象, 在发消息者和 队列之间, 加入了交换器 (Exchange). 

这样发消息者和队列就没有直接联系, 转而变成发消息者把消息给交换器, 交换器根据调度策略再把消息再给队列。

 

• 重要的概念有 4 个，分别为：虚拟主机，交换机，队列，和绑定。
  

虚拟主机：一个虚拟主机持有一组交换机、队列和绑定。为什么需要多个虚拟主机呢？很简单，RabbitMQ当中，用户只能在虚拟主机的粒度进行权限控制。 因此，如果需要禁止A组访问B组的交换机/队列/绑定，必须为A和B分别创建一个虚拟主机。每一个RabbitMQ服务器都有一个默认的虚拟主机“/”。

交换机：Exchange 用于转发消息，但是它不会做存储 ，如果没有 Queue bind 到 Exchange 的话，它会直接丢弃掉 Producer 发送过来的消息。 这里有一个比较重要的概念：路由键 。消息到交换机的时候，交互机会转发到对应的队列中，那么究竟转发到哪个队列，就要根据该路由键。

交换机有四种类型：Direct, topic, Headers and Fanout。

绑定：也就是交换机需要和队列相绑定，这其中如上图所示，是多对多的关系。

 

 

 

• 发送（单次发送）：
  

 

<wiz_code_mirror>

 

 

 

 

 

 

 

        /**

         * 创建连接连接到MabbitMQ

         */

        ConnectionFactory factory = new ConnectionFactory();

        //设置MabbitMQ所在主机ip或者主机名

        factory.setHost("127.0.0.1");

        //创建一个连接

        Connection connection = factory.newConnection();

        //创建一个频道

        Channel channel = connection.createChannel();

        //指定一个队列

        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        //发送的消息

        String message = "hello world!";

        //往队列中发出一条消息

        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());

//channel.basicPublish("", QUEUE_NAME, null, SerializationUtils.serialize(messageInfo));实体使用序列化

        System.out.println("Sent ‘" + message + "‘");

        //关闭频道和连接

        channel.close();

        connection.close();

 

 

 

• 接收（实时接收、循环调用）：

<wiz_code_mirror>

 

 

 

 

 

 

 

//打开连接和创建频道，与发送端一样

        ConnectionFactory factory = new ConnectionFactory();

        //设置MabbitMQ所在主机ip或者主机名

        factory.setHost("127.0.0.1");

        Connection connection = factory.newConnection();

        Channel channel = connection.createChannel();

        //声明队列，主要为了防止消息接收者先运行此程序，队列还不存在时创建队列。

        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        System.out.println("Waiting for messages. To exit press CTRL+C");

?

        //创建队列消费者

        QueueingConsumer consumer = new QueueingConsumer(channel);

        //指定消费队列

        channel.basicConsume(QUEUE_NAME, true, consumer);

        while (true)

        {

            //nextDelivery是一个阻塞方法（内部实现其实是阻塞队列的take方法）

            QueueingConsumer.Delivery delivery = consumer.nextDelivery();

            String message = new String(delivery.getBody());

            //MessageInfo messageInfo = (MessageInfo)SerializationUtils.deserialize(body);实体使用反序列化

            System.out.println("Received ‘" + message + "‘");

        }

 

 

• 配置文件：
  

默认端口5672.

RabbitMQ Server abbitmq_server-3.7.4etc abbitmq.config.example

<wiz_code_mirror>

 

 

 

spring.rabbitmq.host=127.0.0.1

spring.rabbitmq.port=5672

spring.rabbitmq.username=guest

spring.rabbitmq.password=guest