rabbitmq入门示例 - 生产和消费消息（Java）

一时失言乱红尘 2023-06-16 14:56 18阅读 0赞

本文演示RabbitMQ Java中生产和消费消息的示例。

首先在maven配置文件中加入相关依赖：

<dependency>
        <groupId>com.rabbitmq</groupId>
        <artifactId>amqp-client</artifactId>
        <version>4.2.1</version>
    </dependency>

首先启动rabbitmq，我们用的是mac环境，[mac下启动rabbitmq][mac_rabbitmq]。  
默认情况下，访问RabbitMQ服务的用户名和密码都是`guest`，这个账户有限制，默认只能通过本地网络（localhost）访问，远程网络访问受限，所以在生产和消费消息之前要添加一个用户，并设置相应的访问权限。

# 添加新用户
    ➜  ~ git:(master) ✗ rabbitmqctl add_user root 1234
    Adding user "root" ...
    # 为root用户设置所有权限
    ➜  ~ git:(master) ✗ rabbitmqctl set_permissions -p / root ".*" ".*" ".*"
    Setting permissions for user "root" in vhost "/" ...
    # 设置root用户为管理员角色
    ➜  ~ git:(master) ✗ rabbitmqctl set_user_tags root administrator
    Setting tags for user "root" to [administrator] ...

下面开始代码的编写，这里我们实现的是生产者发送一条消息“hello world！”到rabbitmq，之后由消费者消费。

生产者代码：

package com.herobin.scd.test;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.MessageProperties;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /** * @author binzhang * @date 2019-11-26 */
    public class RabbitProducer { 
        private static final String EXCHANGE_NAME = "exchange_demo";
        private static final String ROUTING_KEY = "routingkey_demo";
        private static final String QUEUE_NAME = "queue_demo";
        private static final String IP_ADDRESS = "127.0.0.1";
        private static final int PORT = 5672;//RabbitMQ服务端默认端口号为5672
    
        public static void main(String[] args) throws IOException, TimeoutException, InterruptedException { 
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost(IP_ADDRESS);
            factory.setPort(PORT);
            factory.setUsername("root");
            factory.setPassword("1234");
            Connection connection = factory.newConnection();//创建连接
            Channel channel = connection.createChannel();//创建信道
            // 创建一个type="direct"、持久化的、非自动删除的交换器
            channel.exchangeDeclare(EXCHANGE_NAME, "direct", true, false, null);
            // 创建一个持久化、非排他的、非自动删除的队列
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
            // 发送一条持久化的消息：hello world！
            String message = "hello world!";
            channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
            // 关闭资源
            channel.close();
            connection.close();
        }
    }

这里首先和RabbitMQ建立一个Connection连接，然后在这个连接上创建一个信道Channel。之后创建一个交换器Exchange和一个队列Queue，并通过路由键绑定，然后发送一条消息，最后关闭资源。

在执行程序之前也要在rabbitmq中配置好对应的exchange\_demo、routingkey\_demo和queue\_demo。

执行生产者程序，可以看到rabbitmq中多了一条消息。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9oZXJvYmluLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70]

消费者代码：

package com.herobin.scd.test;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.TimeoutException;
    
    /** * @author binzhang * @date 2019-11-26 */
    public class RabbitConsumer { 
        private static final String QUEUE_NAME = "queue_demo";
        private static final String IP_ADDRESS = "127.0.0.1";
        private static final int PORT = 5672;
    
        public static void main(String[] args) throws IOException, TimeoutException, InterruptedException { 
            Address[] addresses = new Address[] { 
                    new Address(IP_ADDRESS, PORT)
            };
            ConnectionFactory factory = new ConnectionFactory();
            factory.setUsername("root");
            factory.setPassword("1234");
            // 这里连接方式与生产者的demo略有不同，注意辨别区别
            Connection connection = factory.newConnection(addresses);//创建连接
            final Channel channel = connection.createChannel();//创建信道
            channel.basicQos(64);//设置客户端最多接受未被ack的消息的个数
            Consumer consumer = new DefaultConsumer(channel) { 
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    System.out.println("recv message: " + new String(body));
                    try { 
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            };
            channel.basicConsume(QUEUE_NAME, consumer);
            // 等待回调函数执行完毕之后，关闭资源
            TimeUnit.SECONDS.sleep(5);
            channel.close();
            connection.close();
        }
    }

执行程序，执行结果如图成功消费了消息：  
![在这里插入图片描述][20191127113735474.png]  
消息队列中的消息也没有了。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9oZXJvYmluLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70 1]

[mac_rabbitmq]: https://blog.csdn.net/qq_24095055/article/details/88993147
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9oZXJvYmluLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191127113327736.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9oZXJvYmluLmJsb2cuY3Nkbi5uZXQ=,size_16,color_FFFFFF,t_70
[20191127113735474.png]: https://img-blog.csdnimg.cn/20191127113735474.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9oZXJvYmluLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/2019112711464677.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9oZXJvYmluLmJsb2cuY3Nkbi5uZXQ=,size_16,color_FFFFFF,t_70