rabbitMQ入门程序

￡神魔★判官ぃ 2021-09-24 22:10 390阅读 0赞

1.生产者
    /**
     * rabbitMQ入门程序消费者
     *
     * @author xiaoss
     * @date 2020年10月27日 22:02
     */
    public class Producer01 {
    
        //队列
        private static final String QUEUE="helloworld_01";
    
        public static void main(String[] args) {
            //通过连接工厂创建新的连接和mq建立连接
            ConnectionFactory connectionFactory=new ConnectionFactory();
            connectionFactory.setHost("127.0.0.1");//连接主机地址
            connectionFactory.setPort(5672);//设置端口
            connectionFactory.setUsername("guest");
            connectionFactory.setPassword("guest");
            //设置虚拟机，一个mq服务可以设置多个虚拟机，每个虚拟机就相当于一个独立的mq
            connectionFactory.setVirtualHost("/");
    
    
            Connection connection=null;
            try {
                //建立新连接
                connection=connectionFactory.newConnection();
                //创建会话通道，生产者和mq服务 所有的通信都在channel通道中完成
                Channel channel=connection.createChannel();
    
                //监听队列
                //声明队列，如果队列在mq 中没有则要创建
                //参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
                /**
                 * 参数明细
                 * 1、queue 队列名称
                 * 2、durable 是否持久化，如果持久化，mq重启后队列还在
                 * 3、exclusive 是否独占连接，队列只允许在该连接中访问，如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建
                 * 4、autoDelete 自动删除，队列不再使用时是否自动删除此队列，如果将此参数和exclusive参数设置为true就可以实现临时队列（队列不用了就自动删除）
                 * 5、arguments 参数，可以设置一个队列的扩展参数，比如：可设置存活时间
                 */
                channel.queueDeclare(QUEUE,true,false,false,null);
    
                //发送消息
                String message="hello rabbitMQ";
                /**
                 * 参数明细
                 * 1.exchange:交换机，如果不指定就使用默认的mq交换机
                 * 2.routingKey:路由key,交换机根据路由key将消息发送到指定的队列,如果使用默认交换机，路由key设置为队列的名称
                 * 3.props:消息属性
                 * 4.body:消息内容
                 */
                channel.basicPublish("",QUEUE,null,message.getBytes());
                System.out.println("send to mq"+message);
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
    
            }
    
        }
    }

运行main方法，打开[http://localhost:15672/][http_localhost_15672]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW8yOTczMjg_size_16_color_FFFFFF_t_70][]

就会创建一个helloworld\_01的队列，有一条消息等待消费

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW8yOTczMjg_size_16_color_FFFFFF_t_70 1][]

Ready：待消费的消息总数。

Unacked：待应答（待确认）的消息总数。  
Total：总数 Ready+Unacked

2.创建消费者

/**
     * 入门程序消费者
     *
     * @author xiaoss
     * @date 2020年10月27日 23:48
     */
    public class Consumer01 {
    
        //队列
        private static final String QUEUE="helloworld_01";
    
        public static void main(String[] args) {
    
            //通过连接工厂创建新的连接和mq建立连接
            ConnectionFactory connectionFactory=new ConnectionFactory();
            connectionFactory.setHost("127.0.0.1");//连接主机地址
            connectionFactory.setPort(5672);//设置端口
            connectionFactory.setUsername("guest");
            connectionFactory.setPassword("guest");
            //设置虚拟机，一个mq服务可以设置多个虚拟机，每个虚拟机就相当于一个独立的mq
            connectionFactory.setVirtualHost("/");
    
            Connection connection=null;
            try {
                //建立新连接
                connection=connectionFactory.newConnection();
                //创建会话通道，生产者和mq服务 所有的通信都在channel通道中完成
                Channel channel=connection.createChannel();
    
                //监听队列
                //声明队列，如果队列在mq 中没有则要创建
                //参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
                /**
                 * 参数明细
                 * 1、queue 队列名称
                 * 2、durable 是否持久化，如果持久化，mq重启后队列还在
                 * 3、exclusive 是否独占连接，队列只允许在该连接中访问，如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建
                 * 4、autoDelete 自动删除，队列不再使用时是否自动删除此队列，如果将此参数和exclusive参数设置为true就可以实现临时队列（队列不用了就自动删除）
                 * 5、arguments 参数，可以设置一个队列的扩展参数，比如：可设置存活时间
                 */
                channel.queueDeclare(QUEUE,true,false,false,null);
    
                //实现消费方法
                DefaultConsumer defaultConsumer=new DefaultConsumer(channel){
    
                    /**
                     * 当接收到消息后此方法将被调用
                     * @param consumerTag  消费者标签，用来标识消费者的，在监听队列时设置channel.basicConsume
                     * @param envelope 信封，通过envelope
                     * @param properties 消息属性
                     * @param body 消息内容
                     * @throws IOException
                     */
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                        //交换机
                        String exchange = envelope.getExchange();
                        //消息id，mq在channel中用来标识消息的id，可用于确认消息已接收
                        long deliveryTag = envelope.getDeliveryTag();
                        //消息内容
                        String message= new String(body,"utf-8");
                        System.out.println("receive message:"+message);
                    }
                };
    
                //监听队列
                /**
                 * 参数明细
                 * 1.queue:队列名称
                 * 2.autoAck:自动回复，当消费者接收到消息后要告诉mq消息已接收，如果将此参数设置为true表示自动回复mq，设置为false表示要通过编程实现回复
                 * 3.callback:消费方法，当消费者接收到消息要执行的方法
                 */
                channel.basicConsume(QUEUE,true,defaultConsumer);
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
    
            }
        }
    }

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW8yOTczMjg_size_16_color_FFFFFF_t_70 2][]

![20201028001736855.png][]

说明刚才的消息已被消费

[http_localhost_15672]: http://localhost:15672/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW8yOTczMjg_size_16_color_FFFFFF_t_70]: /images/20210923/d0894394757f4994860085ec0f09a0df.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW8yOTczMjg_size_16_color_FFFFFF_t_70 1]: /images/20210923/10bf4d6d89cf4cf2b1f1fc143cb9fd86.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW8yOTczMjg_size_16_color_FFFFFF_t_70 2]: /images/20210923/a259f64341564d57bbf5fe6aac3b9055.png
[20201028001736855.png]: /images/20210923/507943682dab41759396f97358e2f834.png