RabbitMQ学习笔记（二）-----------------RabbitMQ生产消费消息

灰太狼 2021-01-24 18:56 820阅读 0赞

项目地址：[https://github.com/gongxianshengjiadexiaohuihui/RabbitMQ/tree/master/Hello\_RabbitMQ][https_github.com_gongxianshengjiadexiaohuihui_RabbitMQ_tree_master_Hello_RabbitMQ]

项目结构

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTQzNjM0_size_16_color_FFFFFF_t_70][]

需要的jar包

![20181120153821661.png][]

项目流程图

x![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTQzNjM0_size_16_color_FFFFFF_t_70 1][]

首先是生产者的类，我们需要与RabbitServer建立连接，建立通道，并声明一个队列，然后放10条消息进去

package com.ggp;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /**
     * @ClassName Producer
     * @Description 生产者
     * @Author Mr.G
     * @Date 2018/11/20 10:38
     * @Version 1.0
     */
    public class Producer {
        /**
         * 设置队列名
         */
        public final static String QUEUE_NAME = "rabbitMQ.test";
    
        public Producer()throws IOException, TimeoutException {
    
            /**
             * 创建连接工厂
             */
            ConnectionFactory connectionFactory = new ConnectionFactory();
            /**
             * 设置配置信息，如果你装完rabbitmq没有进行配置，默认就是下面的配置
             */
            connectionFactory.setHost("localhost");
            connectionFactory.setUsername("guest");
            connectionFactory.setPassword("guest");
            /**
             * 建立一个新的连接
             */
            Connection connection = connectionFactory.newConnection();
            /**
             * 建立一个通道
             */
            Channel channel = connection.createChannel();
            /**
             * 参数，从左到右
             * 队列名     是否持久化     是否是私有队列  是否自动删除    map
             */
            channel.queueDeclare(QUEUE_NAME,false,false,false,null);
            /**
             *
             */
            for(int i = 0; i < 10; i++) {
                String message = "Hello rabbitmq !" + i;
    
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));
    
                System.out.println("Producer Send  Message: " + message);
            }
    
            channel.close();
    
            connection.close();
        }
    
    }

然后创建消费者的类，同样是建立连接，通道，然后监听这个通道，如果有信息的话，就取出处理，每次设定取出一条，处理时间设为1秒

package com.ggp;
    
    
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    
    import java.util.concurrent.TimeoutException;
    
    /**
     * @ClassName Customer
     * @Description TODO
     * @Author Mr.G
     * @Date 2018/11/20 10:57
     * @Version 1.0
     */
    public class Customer {
    
        private final static String QUEUE_NAME = "rabbitMQ.test";
    
        private String cumstomerName;
    
        public Customer(final String customerName) throws IOException, TimeoutException {
    
            this.cumstomerName = customerName;
    
            ConnectionFactory connectionFactory = new ConnectionFactory();
    
            connectionFactory.setHost("localhost");
            connectionFactory.setUsername("guest");
            connectionFactory.setPassword("guest");
    
            Connection connection = connectionFactory.newConnection();
            Channel channel = connection.createChannel();
            channel.queueDeclare(QUEUE_NAME,false,false,false,null);
    
            System.out.println(customerName+" Waiting Received messages");
            /**
             * 每次从队列种获取消息的数量
             */
            channel.basicQos(1);
            /**DefaultConsumer类实现了Consumer接口，通过传入一个通道，告诉服务我们需要那个通道的信息，如果通道有信息传回，就会执行回调函数handldDelivery
             *
             */
            Consumer consumer = new DefaultConsumer(channel){
              @Override
              /**
               * 消息分发
               */
              public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)throws IOException{
                  String message = new String(body, "UTF-8");
                  System.out.println(customerName+" Received Message: "+message);
                  System.out.println(customerName+" begin dealing");
                  try {
                      Thread.sleep(1000);
                  } catch (InterruptedException e) {
                      Thread.currentThread().interrupt();
                  }
                  System.out.println(customerName+" dealing successfully!");
              }
            };
    
            /**
             * 自动回复队列应答    RabbitMQ的消息确认机制
             * 第二个参数为自动回复
             */
             channel.basicConsume(QUEUE_NAME,true,consumer);
    
        }
    
    }

最后是我们测试类

package com.ggp.test;
    
    import com.ggp.Customer;
    import com.ggp.Producer;
    
    /**
     * @ClassName DemoTest
     * @Description TODO
     * @Author Mr.G
     * @Date 2018/11/20 15:08
     * @Version 1.0
     */
    public class DemoTest {
        public static void main(String[] args) throws Exception{
    
            Customer customer1 = new Customer("GGP");
            Customer customer2 = new Customer("RQB");
            new Producer();
        }
    }

运行结果

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTQzNjM0_size_16_color_FFFFFF_t_70 2][]

[https_github.com_gongxianshengjiadexiaohuihui_RabbitMQ_tree_master_Hello_RabbitMQ]: https://github.com/gongxianshengjiadexiaohuihui/RabbitMQ/tree/master/Hello_RabbitMQ
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTQzNjM0_size_16_color_FFFFFF_t_70]: /images/1611514583188.png
[20181120153821661.png]: /images/1611514560112.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTQzNjM0_size_16_color_FFFFFF_t_70 1]: /images/1611514540861.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzNTQzNjM0_size_16_color_FFFFFF_t_70 2]: /images/1611514520631.png