RabbitMQ消息队列之二:消费者和生产者 Demo

本是古典何须时尚 2022-06-09 03:45 192阅读 0赞

在使用RabbitMQ之前，需要了解RabbitMQ的工作原理。

## RabbitMQ的工作原理 ##

RabbitMQ是消息代理。从本质上说，它接受来自生产者的信息，并将它们传递给消费者。在两者之间，它可以根据你给它的路由，缓冲规则进行传递消息。

## 示例图 ##

![这里写图片描述][SouthEast]

## RabbitMQ的术语解释 ##

生产者：生产消息，发送消息。类似工厂。  
消费者：接受消息，使用消息。类似顾客。  
队列：存储消息。类似仓库、中转站。队列可以存储很多的消息，因为它基本上是一个无限制的缓冲区，前提是你的机器有足够的存储空间。多个生产者可以将消息发送到同一个队列中，多个消费者也可以只从同一个队列接收数据。这就是队列的特性。

## RabbitMQ 消息生产者的代码实现 ##

import java.util.HashMap;
    import java.util.Map;
    import com.alibaba.fastjson.JSON;
    import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.Channel;
    public class RabbitProducer { 
        private final static String QUEUE_NAME = "RabbitMQ_Hello"; //消息队列名
    
          public static void main(String[] argv) throws Exception {
             //创建连接连接到RabbitMQ 
            ConnectionFactory factory = new ConnectionFactory();
            // 设置ip
            factory.setHost("127.0.0.1");
            /* //设置端口 factory.setPort(15672); //设置用户名 factory.setUsername("guest"); //设置密码 factory.setPassword("guest"); //设置url(包括ip、端口、用户名、密码) factory.setUri("amqp://guest:guest@localhost:15672"); */    
            // 创建一个连接 
            Connection connection = factory.newConnection();
            // 创建一个频道 
            Channel channel = connection.createChannel();
            // 指定一个队列 
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            Map<String,Object> map=new HashMap<String,Object>();  
              map.put("java", "hello");
              map.put("RabbitMQ", "Hello");
            //发送的消息
            String message = JSON.toJSONString(map); 
            // 往队列中发出一条消息 
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes()); 
            System.out.println(" [x] Sent '" + message + "'");
            // 关闭频道和连接 
            channel.close();
            connection.close();   
          }
    }

## RabbitMQ消息消费者的代码实现 ##

import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.QueueingConsumer;
    
    //消费者 
    public class RabbitConsumer { 
    
      private final static String QUEUE_NAME = "RabbitMQ_Hello"; //消息队列名
    
      public static void main(String[] argv) throws Exception {
    
            ConnectionFactory factory = new ConnectionFactory();
            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) {  //消费者程序运行开着 如果生产者新增了数据会自动获取
              Thread.sleep(500);
                 // nextDelivery是一个阻塞方法（内部实现其实是阻塞队列的take方法） 
              QueueingConsumer.Delivery delivery = consumer.nextDelivery();
              String message = new String(delivery.getBody());
              System.out.println("'[x] Received '" + message );
      }   
      }
    }

打开本地的RabbitMQ，之后运行生产者代码，会将消息存储到RabbitMQ中，可以通过RabbitMQ web界面看到消息。

![这里写图片描述][SouthEast 1]

![这里写图片描述][SouthEast 2]

![这里写图片描述][SouthEast 3]

![这里写图片描述][SouthEast 4]

运行消费者代码，会将该消息消费掉，可以在RabbitMQ web界面可以发现上面存储的消息已经没有了。  
![这里写图片描述][SouthEast 5]

![这里写图片描述][SouthEast 6]

通过这个demo，可以对消息队列有了一个的基本的了解了。

[SouthEast]: /images/20220609/50af9e30cb464056ad9874d84e67a6dc.png
[SouthEast 1]: /images/20220609/47ab0a5853c948f8b7ac2786f3192e02.png
[SouthEast 2]: /images/20220609/69d85cc75003412eb549f469d41651e6.png
[SouthEast 3]: /images/20220609/6101c4e4e46740029e87be6d10b66d83.png
[SouthEast 4]: /images/20220609/7455174c401d424a8e16ea42f12257af.png
[SouthEast 5]: /images/20220609/b3dd4893c3fc42a28746db15bb2629cf.png
[SouthEast 6]: /images/20220609/6d0798b66ec74ed1aec56d00efaba7c2.png