RabbitMQ

rabbitmq-plugins enable rabbitmq_management //开启Web插件

rabbitmq-server start //启动服务

rabbitmq-server stop //停止服务

rabbitmq-server restart //重启服务

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

1. 创建连接

   //创建一个连接工厂
           ConnectionFactory factory = new ConnectionFactory();
           //服务地址
           factory.setHost("localhost");
           //账号
           factory.setUsername("guest");
           //密码
           factory.setPassword("guest");
           //端口号
           factory.setPort(5672);
           //创建连接
           Connection connection = factory.newConnection();
           //创建信道
           Channel channel = connection.createChannel();

2. 创建交换机

   /**
            * 创建交换机
            * 1，交换机名称
            * 2.交换机类型，direct,topic,fanout,headers
            * 3.指定交换机是否需要持久化，如果设置为true，那么交换机的元数据需要持久化
            * 4.指定交换机在没有队列绑定时，是否需要删除，设置false表示不删除
            * 5.Map<String,Object>类型，用来指定交换机其他的一些机构化参数，一般直接设置为null
            */
           channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT,true,false,null);

3. 生成一个队列

   /**
            * 生成一个队列
            * 1.队列名称
            * 2.队列是否需要持久化，但是要注意，只是队列名称等这些元数据的持久化，不是队列中消息的持久化
            * 3.表示队列是不是私有化，如果是私有化的，只有创建他的应用程序才能消费消息
            * 4.队列在没有消费者订阅的情况下是否自动删除；
            * 5.队列的一些结构化信息，比如声明死信队列，磁盘队列会用到；
            */
           channel.queueDeclare(queueName,true,false,false,null);

4. 将我们的交换机和队列绑定

   /**
            * 将我们的交换机和队列绑定
            * 1.队列名称
            * 2.交换机名称
            * 3.路由键，在直连模式下，可以为我们的队列名称
            */
           channel.queueBind(queueName,exchangeName,queueName);

5. 发送消息

           String queueName ="xc_queue_name";
           String exchangeName ="xc_exchange_name";
        
    　　   //发送消息
           String message = "hello rabbitmq";
           System.out.println("消息发送成功");
           /**
            * 发送消息
            * 1.发送到哪个交换机
            * 2.队列名称
            * 3.其他参数信息
            * 4,发送消息的消息体
            */
           channel.basicPublish(exchangeName,queueName,null,message.getBytes());

6. 关闭连接和通道

           channel.close();
           connection.close();   

    

1. 建立连接

   //创建一个连接工厂
           ConnectionFactory factory = new ConnectionFactory();
           //服务地址
           factory.setHost("localhost");
           //账号
           factory.setUsername("guest");
           //密码
           factory.setPassword("guest");
           //端口号
           factory.setPort(5672);
           //创建连接
           Connection connection = factory.newConnection();
           //创建信道
           Channel channel = connection.createChannel();

2. 回调函数

   //接受消息的回调函数
           DeliverCallback deliverCallback = (consumerTage,message) ->{
               System.out.println("接收到消息"+new String(message.getBody()));
           };
           //取消消息的回调函数
           CancelCallback cancelCallback = consumerTage ->{
               System.out.println("消费信息被中断");
           };

3. 消费消息

   /**
            * 消费消息
            * 1.消费哪个队列
            * 2.消费成功之后是否需要自动应答，true 自动应答
            * 3.接受消息的回调函数
            * 4.取消消息的回调
            */
           channel.basicConsume("xc_queue_name",true,deliverCallback,cancelCallback);

    

1. topic中，将routingkey通过“.”来分为多个部分；
2. “*”：代表一个部分；
3. “#”：代表0个或多个部分（如果绑定的路由键为“#”时，则接受所有消息，因为路由键所有都匹配）；

   

1. x-match = all：表示所有的键值对都匹配才能接受到消息；
2. x-match = any：表示只要有键值对匹配就能接受到消息；