RabbitMQ小试牛刀

小鱼儿 2022-05-13 00:56 213阅读 0赞

1.安装Erlang语言，和RabbitMQ server（可以选择windows和linux系统）（这个可以百度），我用的windows环境。

进入../sbin (rabbitMQ安装目录)

rabbitmqctl -q status       //打印了一些rabbitmq服务状态信息，包括内存，硬盘，和使用erlong的版本信息

rabbitmqctl list\_queues     //查看所有队列消息

2. 默认情况下RabbitMQ服务的用户名和密码都是“guest”，这个账户是有限制的，默认只能通过localhost访问，远程访问权限受限，所以需要给它添加一个用户，并设置相应的访问权限。

> 添加一个新用户 "root"，密码是“root”:
> 
>     E:\RabbitMQ\rabbitmq_server-3.7.7\sbin>rabbitmqctl add_user root root
>     Creating  user  " root "  ....
> 
> 为root用户设置权限:
> 
>     E:\RabbitMQ\rabbitmq_server-3.7.7\sbin>rabbitmqctl set_permission -p / root ".*"  ".*"  ".*"
>     Setting permissions for user  "root " in vhost "/" ...
> 
> 设置root用户为管理员角色:
> 
>     E:\RabbitMQ\rabbitmq_server-3.7.7\sbin>rabbitmqctl set_user_tags root administrator
>     Setting  tags  for  user  " root"  to  [administrator] ...
> 
>

3.新建maven项目，导入依赖

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

4.客户端消息生产者代码

package com.wusu.consumer;
    
    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;
    
    public class RabbitmqProducer {
        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;
    
        public static void main(String[] args) throws IOException, TimeoutException {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost(IP_ADDRESS);
            factory.setPort(PORT);
            factory.setUsername("root");
            factory.setPassword("root");
            Connection connection = factory.newConnection(); //创建连接
            Channel channel = connection.createChannel();//创建信道
            //创建一个type="direct" 、持久化，非自动删除的交换器
            channel.exchangeDeclare(EXCHANGE_NAME,"direct",true,false,null);
            //创建一个持久化，非排他的，非自动删除的队列
            channel.queueDeclare(QUEUE_NAME,true,false,false,null);
            //将交换器与队列通过路由绑定
            channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,ROUTING_KEY);
            //发送持久化信息
            String message = "hello world";
            channel.basicPublish(EXCHANGE_NAME,ROUTING_KEY,MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());
            channel.close();
            connection.close();
        }
    }

5.客户端消息消费者代码

package com.wusu.producer;
    
    import com.rabbitmq.client.*;
    import org.omg.CORBA.TIMEOUT;
    
    import java.awt.*;
    import java.io.IOException;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.TimeoutException;
    
    public class RabbitmqConsumer {
        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(IP_ADDRESS, PORT)};
            ConnectionFactory factory = new ConnectionFactory();
            factory.setPassword("root");
            factory.setUsername("root");
            Connection connection = factory.newConnection(addresses); //创建连接
            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("recieve 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();
        }
    
    }

6.通过rabbitMQ控制台查看消息情况

首先进入安装目录的.../sbin中，使用如下命令开启网页版控制台

E:\RabbitMQ\rabbitmq_server-3.7.7\sbin>rabbitmq-plugins enable rabbitmq_management

然后本地使用[http://127.0.0.1:15672][http_127.0.0.1_15672]   即可打开RabbitMQ的控制台

如下所示，生产者生产了一条消息：

![70][]

![70 1][]

当本地消费者把这条消息消费后，里面的消息就没有了。

[http_127.0.0.1_15672]: http://127.0.0.1:15672
[70]: /images/20220513/69a33b03cee1418a8f7b9bdd86627f53.png
[70 1]: /images/20220513/fcb2d32df4ed4ee5b0b02fd24f812424.png