正文

Message Queue（消息队列），从字⾯上理解：⾸先它是⼀个队列。 FIFO 先进先出的数据结构—— 队列。消息队列就是所谓的存放消息的队列。消息队列解决的不是存放消息的队列的⽬的，解决的是通信问题。⽐如以电商订单系统为例，如果各服务之间使⽤同步通信，不仅耗时较久且过程中受到⽹络波动的影响，不能保证⾼成功率。因此，使⽤异步的通信⽅式对架构进⾏改造。使⽤异步的通信⽅式对模块间的调⽤进⾏解耦，可以快速的提升系统的吞吐量。上游执⾏完消息的发送业务后⽴即获得结果，下游多个服务订阅到消息后各⾃消费。通过消息队列，屏蔽底层的通信协议，使得解藕和并⾏消费得以实现。

二、RabbitMQ介绍

市⾯上⽐较⽕爆的⼏款 MQ ： ActiveMQ ， RocketMQ ， Kafka ， RabbitMQ 。

语⾔的⽀持：ActiveMQ，RocketMQ只⽀持Java语⾔，Kafka可以⽀持多们语⾔，RabbitMQ⽀持多种语⾔。
效率⽅⾯：ActiveMQ，RocketMQ，Kafka效率都是毫秒级别，RabbitMQ是微秒级别的。
消息丢失，消息重复问题： RabbitMQ针对消息的持久化，和重复问题都有⽐较成熟的解决⽅案。
学习成本：RabbitMQ⾮常简单。
官网：Messaging that just works — RabbitMQ

RabbitMQ 是由 Rabbit 公司去研发和维护的，最终是在 Pivotal。 RabbitMQ 严格的遵循 AMQP 协议，⾼级消息队列协议，帮助我们在进程之间传递异步消息。

三、RabbitMQ安装

环境：ubuntu、docker、docker-compose

编写docker-compose.yml

version: "3.1"
services:
  rabbitmq:
    image: daocloud.io/library/rabbitmq:management
    restart: always
    container_name: rabbitmq
    ports:
      - 5672:5672
      - 15672:15672
    volumes:
      - ./data:/var/lib/rabbitmq

执行

docker-compose up -d

四、RabbitMQ架构

1.官⽅的简单架构图

Publisher - ⽣产者：发布消息到 RabbitMQ 中的 Exchange Consumer - 消费者：监听 RabbitMQ 中的 Queue 中的消息 Exchange - 交换机：和⽣产者建⽴连接并接收⽣产者的消息 Queue - 队列： Exchange 会将消息分发到指定的 Queue ， Queue 和消费者进⾏交互 Routes - 路由：交换机以什么样的策略将消息发布到 Queue

2.RabbitMQ的完整架构图

3.查看图形化界⾯并创建⼀个Virtual Host

虚拟主机就是⽤来将⼀个 rabbitmq 内部划分成多个主机，给不同的⽤户来使⽤，⽽不会冲突。

创建⼀个全新的⽤户和全新的 Virtual Host ，并且将 test ⽤户设置上可以操作/test 的权限

五、RabbitMQ的队列模式

1.RabbitMQ的通讯⽅式

地址：RabbitMQ Tutorials — RabbitMQ

2.HelloWorld模式-简单队列模式

1 ）创建消息的⽣产者

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

package com.wang.helloword;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;


/**
 * @BelongsProject: RabbitMqLearn
 * @BelongsPackage: com.wang
 * @Author: wang fei
 * @CreateTime: 2023-02-03  15:58
 * @Description: TODO rabbitmq-helloword队列模式（简单队列模式）生产者
 * @Version: 1.0
 */
public class MyProducer 
    public static final String QUEUE_NAME = "my_queue";

    public static void main(String[] args) throws IOException, TimeoutException 
        //获得连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.206.130");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("wang");
        connectionFactory.setPassword("123456");
        connectionFactory.setVirtualHost("/test1");
        //获得链接对像
        Connection connection = connectionFactory.newConnection();
        //获得信道
        Channel channel = connection.createChannel();
        //声明队列
    /*
queue – the name of the queue 队列名称
durable – true if we are declaring a durable queue (the queue will
survive a server restart) 队列是否持久化
exclusive – true if we are declaring an exclusive queue
(restricted to this connection) 独占：是否只给当前客户端来使⽤
autoDelete – true if we are declaring an autodelete queue (server
will delete it when no longer in use) ⾃动删除
arguments – other properties (construction arguments) for the queue
 */
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //定义消息
        String message = "我是生产者";
        //发送消息
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println("消息发送成功");
        //断开链接
        channel.close();
        connection.close();

2 ）创建消息的消费者

package com.wang.helloword;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @BelongsProject: RabbitMqLearn
 * @BelongsPackage: com.wang
 * @Author: wang fei
 * @CreateTime: 2023-02-03  16:49
 * @Description: TODORabbitMq helloword消费者
 * @Version: 1.0
 */
public class MyConsumer 
    public static final String QUEUE_NAME ="my_queue";

    public static void main(String[] args) throws IOException, TimeoutException 
        //获取链接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.206.130");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("wang");
        connectionFactory.setPassword("123456");
        connectionFactory.setVirtualHost("/test1");
        //获得链接对像
        Connection connection = connectionFactory.newConnection();
        //获得信道
        Channel channel = connection.createChannel();
        //使⽤⾮Lambda的⽅式来消费
//        //创建Consumer对象,指明具体的消息处理程序
//        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(message);
//            
//        ;
//        //设置消费者监听queue("my_queue")
//        channel.basicConsume(QUEUE_NAME, true, consumer);

        //Lambda的⽅式来消费
        DeliverCallback deliverCallback = (consumerTag, delivery) -> 
            //处理信息,并且进行业务逻辑处理,打印信息
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(message);
        ;
        //设置消费者监听queue("my_queue")
        channel.basicConsume(QUEUE_NAME,true,deliverCallback,consumerTag->);

简单队列的问题： 当多个消费者消费同⼀个队列时。这个时候 rabbitmq 的公平调度机制就开启了，于是，⽆论消费者的消费能⼒如何，每个消费者都能公平均分到相同数量的消息，⽽不能出现能者多劳的情况。

3. work 队列模式：能者多劳模式

将⾃动ack 改成⼿动 ack：消费者先声明⼀次只接收⼀条消息： channel.basicQos(1) 消费者关闭⾃动 ack 消费者消费完消息后⼿动 ack 1）创建消费者

package com.wang.work;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @BelongsProject: RabbitMqLearn
 * @BelongsPackage: com.wang
 * @Author: wang fei
 * @CreateTime: 2023-02-03  16:49
 * @Description: TODO RabbitMq work队列模式（能者多劳）消费者,消费者先声明⼀次只接收⼀条消息： channel.basicQos(1),消费者关闭⾃动ack
 * @Version: 1.0
 */
public class MyConsumer 
    public static final String QUEUE_NAME ="my_work_queue";

    public static void main(String[] args) throws IOException, TimeoutException 
        //获取链接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.206.130");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("wang");
        connectionFactory.setPassword("123456");
        connectionFactory.setVirtualHost("/test1");
        //获得链接对像
        Connection connection = connectionFactory.newConnection();
        //获得信道
        Channel channel = connection.createChannel();
        //表示⼀次只接收⼀条消息
        channel.basicQos(1);

       //创建Consumer对象,指明具体的消息处理程序
        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(message);
                //⼿动ack 传的是消息的Tag标记，⽤来表示当前处理的这条消息
                channel.basicAck(envelope.getDeliveryTag(),true);
            
        ;
        //设置消费者监听queue("my_queue"),把⾃动ack改为⼿动ack
        channel.basicConsume(QUEUE_NAME, false, consumer);

2）创建信息⽣产者

package com.wang.work;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;


/**
 * @BelongsProject: RabbitMqLearn
 * @BelongsPackage: com.wang
 * @Author: wang fei
 * @CreateTime: 2023-02-03  15:58
 * @Description: TODO TODORabbitMq work队列模式（能者多劳）生产者
 * @Version: 1.0
 */
public class MyProducer 
    public static final String QUEUE_NAME ="my_work_queue";

    public static void main(String[] args) throws IOException, TimeoutException 
        //获得连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.206.130");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("wang");
        connectionFactory.setPassword("123456");
        connectionFactory.setVirtualHost("/test1");
        //获得链接对像
        Connection connection = connectionFactory.newConnection();
        //获得信道
        Channel channel = connection.createChannel();
        //声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        //发送消息
        for (int i = 0; i < 100; i++) 
            //定义消息
            String message = "我是生产者"+i;
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println("消息发送成功");
        
        //断开链接
        channel.close();
        connection.close();

4. 发布订阅模式-fanout

对于之前的队列模式，是没有办法解决⼀条消息同时被多个消费者消费。于是使⽤发布订阅模式来实现。 1 ）编写⽣产者 关键步骤：声明交换机、把消息发送到交换机上

package com.wang.pubsub;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.wang.utils.RabbitMqUtil;




/**
 * @BelongsProject: RabbitMqLearn
 * @BelongsPackage: com.wang
 * @Author: wang fei
 * @CreateTime: 2023-02-03  15:58
 * @Description: TODO rabbitmq-fanout队列模式生产者
 * @Version: 1.0
 */
public class MyProducer 
    //定义交换机名称
    public  static String EXCHANGE_NAME = "my_fanout_exchange";

    public static void main(String[] args) throws Exception 
        Connection connection=RabbitMqUtil.getConnection();
        Channel channel = connection.createChannel();
        //声名交换机
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        //发送信息
        for (int i = 0; i < 20; i++) 
            String message = "Hello RabbitMq!";
            channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
        
        System.out.println("消息发送成功");
        //断开链接
        channel.close();
        connection.close();

2 ）编写消费者 关键动作：

创建队列
创建交换机
把队列绑定在交换机上
让消费者监听队列

package com.wang.pubsub;

import com.rabbitmq.client.*;
import com.wang.utils.RabbitMqUtil;

import java.io.IOException;


/**
 * @BelongsProject: RabbitMqLearn
 * @BelongsPackage: com.wang
 * @Author: wang fei
 * @CreateTime: 2023-02-03  15:58
 * @Description: TODO 发布订阅模式-fanout 对于之前的队列模式，是没有办法解决⼀条消息同时被多个消费者消费。于是使⽤发布订阅模式来实现。
 * @Version: 1.0
 */
public class MyConsumer
    //发布订阅模式
    //定义交换机名称
    public  static String EXCHANGE_NAME = "my_fanout_exchange";

    //定义队列名称
    public  static String QUEUE_NAME = "my_fanout_queue";
    
    public static void main(String[] args) throws Exception 
        Connection connection = RabbitMqUtil.getConnection();
        Channel channel = connection.createChannel();
        //声名队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //声名交换机
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        //将队列绑定到交换机上
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");

        Consumer consumer = new DefaultConsumer(channel) 
            @Override
            public void handleDelivery(String s, Envelope envelope, AMQP.BasicProperties basicProperties, byte[] bytes) throws IOException 
                System.out.println("接收到消息为:" + new String(bytes));
            
        ;

        //监听队列
        channel.basicConsume(QUEUE_NAME, true,consumer);

5.routing模式-direct

关键动作：

在⽣产者发送消息时指明routing-key
在消费者声明队列和交换机的绑定关系时，指明routing-key

1 ）编写⽣产者

package com.wang.routing;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.wang.utils.RabbitMqUtil;




/**
 * @BelongsProject: RabbitMqLearn
 * @BelongsPackage: com.wang
 * @Author: wang fei
 * @CreateTime: 2023-02-03  15:58
 * @Description: TODO rabbitmq-routing模式-direct 生产者
 * 关键动作：
 * 在⽣产者发送消息时指明routing-key
 * 在消费者声明队列和交换机的绑定关系时，指明routing-key
 * @Version: 1.0
 */
public class MyProducer 
    //定义交换机名称
    public  static String EXCHANGE_NAME = "my_routing_exchange";

    public static void main(String[] args) throws Exception 
        Connection connection=RabbitMqUtil.getConnection();
        Channel channel = connection.createChannel();
        //声名交换机
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
        //发送信息
        for (int i = 0; i < 20; i++) 
            String message = "Hello routing模式-direct!";
            channel.basicPublish(EXCHANGE_NAME, "my_routing_key", null, message.getBytes());
        
        System.out.println("消息发送成功");
        //断开链接
        channel.close();
        connection.close();

2 ）编写消费者

package com.wang.routing;

import com.rabbitmq.client.*;
import com.wang.utils.RabbitMqUtil;

import java.io.IOException;


/**
 * @BelongsProject: RabbitMqLearn
 * @BelongsPackage: com.wang
 * @Author: wang fei
 * @CreateTime: 2023-02-03  15:58
 * @Description: TODO routing模式-direct。 消费者
 * @Version: 1.0
 */
public class MyConsumer 
    //发布订阅模式
    //定义交换机名称
    public  static String EXCHANGE_NAME = "my_routing_exchange";

    //定义队列名称
    public  static String QUEUE_NAME = "my_routing_queue";
    //定义路由键名称
    public  static String ROUTING_KEY = "my_routing_key";


    public static void main(String[] args) throws Exception 
        Connection connection = RabbitMqUtil.getConnection();
        Channel channel = connection.createChannel();
        //声名队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //声名交换机
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
        //将队列绑定到交换机上
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);

        Consumer consumer = new DefaultConsumer(channel) 
            @Override
            public void handleDelivery(String s, Envelope envelope, AMQP.BasicProperties basicProperties, byte[] bytes) throws IOException 
                System.out.println("接收到消息为:" + new String(bytes));
            
        ;

        //监听队列
        channel.basicConsume(QUEUE_NAME, false,consumer);

6.topics模式

在 routing 模式的基础上，对 routing-key 使⽤了通配符，提⾼了匹配的范围，增加了可玩性。

绑定关系中如果使⽤了 product.* , 那么在发送消息时： product.add ok product.del ok product.add.one 不 ok 绑定关系中如果使⽤了 product.#, 那么在发送消息时： product.add ok product.add.one ok

1 ）编写⽣产者

package com.wang.topic;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.wang.utils.RabbitMqUtil;




/**
 * @BelongsProject: RabbitMqLearn
 * @BelongsPackage: com.wang
 * @Author: wang fei
 * @CreateTime: 2023-02-03  15:58
 * @Description: TODO rabbitmq-topics模式 生产者
 * 绑定关系中如果使⽤了product.* ,那么在发送消息时：
 * product.add ok
 * product.del ok
 * product.add.one 不ok
 *
 *绑定关系中如果使⽤了product.#,那么在发送消息时：
 * product.add ok
 * product.add.one ok
 * @Version: 1.0
 */
public class MyProducer 
    //定义交换机名称
    public  static String EXCHANGE_NAME = "my_topic_exchange";

    public static void main(String[] args) throws Exception 
        Connection connection=RabbitMqUtil.getConnection();
        Channel channel = connection.createChannel();
        //声名交换机
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");
        //发送信息
        String message = "topics模式!";
        channel.basicPublish(EXCHANGE_NAME, "product.add.one", null, message.getBytes());
        System.out.println("消息发送成功");
        //断开链接
        channel.close();
        connection.close();

2 ）编写消费者

package com.wang.topic;


import com.rabbitmq.client.*;
import com.wang.utils.RabbitMqUtil;

import java.io.IOException;


/**
 * @author 飞
 */
public class MyConsumer1 
    //交换机的名称
    private static String EXCHANGE_NAME = "my_topic_exchange";
    //队列的名称
    private static String QUEUE_NAME = "my_topic_queue_1";

    public static void main(String[] args) throws Exception 
        Connection connection = RabbitMqUtil.getConnection();
        Channel channel = connection.createChannel();
        //声明交换机
        channel.exchangeDeclare(EXCHANGE_NAME,"topic");
        //声明队列
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //绑定
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"product.*");
        //创建消费者
        Consumer consumer = new DefaultConsumer(channel)
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException 
                System.out.println("product.* 消费者："+new String(body));
            
        ;
        //让消费者监听队列
        channel.basicConsume(QUEUE_NAME,consumer);

工具类：

package com.wang.utils;

import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

/**
 * @BelongsProject: RabbitMqLearn
 * @BelongsPackage: com.wang.utils
 * @Author: wang fei
 * @CreateTime: 2023-02-03  18:07
 * @Description: TODO 工具类
 * @Version: 1.0
 */
public class RabbitMqUtil 
    private static ConnectionFactory connectionFactory;
    private static String USER_NAME;
    private static String PASSWORD;
    private static String VIRTUAL_HOST;
    private static String HOST;
    private static int PORT;

    static 
        USER_NAME = "wang";
        PASSWORD = "123456";
        HOST = "192.168.206.130";
        PORT = 5672;
        VIRTUAL_HOST = "/test1";
        connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(HOST);
        connectionFactory.setPort(PORT);
        connectionFactory.setUsername(USER_NAME);
        connectionFactory.setPassword(PASSWORD);
        connectionFactory.setVirtualHost(VIRTUAL_HOST);
    

    /**
     * @description: 获得连接对象
     * @method: getConnection
     * @author: wang fei
     * @date: 2023/2/3 18:17:02
     * @param: []
     * @return: com.rabbitmq.client.Connection
     **/
    public static Connection getConnection() throws Exception 
        return connectionFactory.newConnection();

六、在Springboot中使⽤RabbitMQ

准备 1）. 引⼊依赖

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>

2）. 编写配置⽂件

server:
  port: 8090
spring:
  application:
    name: my-spring-boot-producer
  rabbitmq:
    port: 5672
    host: 192.168.206.130
    username: wang
    password: 123456
    virtual-host:  /test1

1.使⽤发布订阅模式

编写配置类

package com.wang.config;

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @BelongsProject: RabbitMqLearn
 * @BelongsPackage: com.wang.com.wang.config
 * @Author: wang fei
 * @CreateTime: 2023-02-04  17:08
 * @Description: TODO 声明队列、交换机、绑定关系
 * @Version: 1.0
 */
@Configuration
public class RabbitConfig 


    private static String QUEUE_NAME ="my_boot_fanout_queue";
    private static String EXCHANGE_NAME = "my_boot_fanout_exchange";

    /**
     * @description: 声明队列
     * @method: queue
     * @author: wang fei
     * @date: 2023/2/4 17:14:03
     * @param: []
     * @return: org.springframework.amqp.core.Queue
    **/
    @Bean
    public Queue queue()
        return  new Queue(QUEUE_NAME,true,false,false);
    

    /**
     * @description: 声明交换机
     * @method: exchange
     * @author: wang fei
     * @date: 2023/2/4 17:14:54
     * @param: []
     * @return: org.springframework.amqp.core.FanoutExchange
    **/
    @Bean
    public FanoutExchange exchange()
        return   new FanoutExchange(EXCHANGE_NAME,true,false);
    

    /**
     * @description: 声明绑定关系 注意：后⾯多了with携带routing-key
     * @method: binding
     * @author: wang fei
     * @date: 2023/2/4 17:17:04
     * @param: [queue, exchange]
     * @return: org.springframework.amqp.core.Binding
    **/
    @Bean
    public Binding binding(Queue queue, FanoutExchange exchange) 
        return BindingBuilder.bind(queue).to(exchange);

编写消费消息的⽅法

关键：使⽤该注解来指定监听的队列 @RabbitListener(queues = "my_boot_fanout_queue")

package com.wang.consumer;

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * @BelongsProject: RabbitMqLearn
 * @BelongsPackage: com.wang.config.consumer
 * @Author: wang fei
 * @CreateTime: 2023-02-04  17:20
 * @Description: TODO MQ 订阅者
 * @Version: 1.0
 */
@Component
public class MyConsumer 

    /**
     * @description: 监听队列，当队列中有消息的时候，该⽅法会被回调，⽤来消费消息
     * @method: receive
     * @author: wang fei
     * @date: 2023/2/4 17:22:35
     * @param: [message]
     * @return: void
    **/
    @RabbitListener(queues = "my_boot_fanout_queue")
    public void receive(String message)
        byte[] msg = message.getBytes();
        System.out.println("收到消息：" + new String(msg));

2 ）编写⽣产者

编写配置类

package com.wang.config;

import org.springframework.amqp.core.FanoutExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @BelongsProject: RabbitMqLearn
 * @BelongsPackage: com.wang.config
 * @Author: wang fei
 * @CreateTime: 2023-02-04  17:30
 * @Description: TODO
 * @Version: 1.0
 */
@Configuration
public class RabbitConfig 
    private static String EXCHANGE_NAME = "my_boot_fanout_exchange";

    /**
     * 声明交换机
     */
    @Bean
    public FanoutExchange exchange()
        return new FanoutExchange(EXCHANGE_NAME,true,false);

使⽤RabbitTemplate发送消息

  @Autowired
    RabbitTemplate rabbitTemplate;

    @Test
    void contextLoads() 
    String msg="Hello spring boot rabbit mq";
    rabbitTemplate.convertAndSend("my_boot_fanout_exchange","", msg);

2.使⽤topic模式

topic 模式相⽐发布订阅模式，多了 routing-key 的使⽤

调整消费者配置类

package com.example.config;


import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @BelongsProject: RabbitMqLearn
 * @BelongsPackage: com.wang.com.wang.config
 * @Author: wang fei
 * @CreateTime: 2023-02-04  17:08
 * @Description: TODO 声明队列、交换机、绑定关系
 * @Version: 1.0
 */
@Configuration
public class RabbitConfig 


    private static String QUEUE_NAME ="my_boot_potic_queue";
    private static String EXCHANGE_NAME = "my_boot_topic_exchange";

    /**
     * @description: 声明队列
     * @method: queue
     * @author: wang fei
     * @date: 2023/2/4 17:14:03
     * @param: []
     * @return: org.springframework.amqp.core.Queue
    **/
    @Bean
    public Queue queue()
        return  new Queue(QUEUE_NAME,true,false,false);
    

    /**
     * @description: 声明交换机
     * @method: exchange
     * @author: wang fei
     * @date: 2023/2/4 17:14:54
     * @param: []
     * @return: org.springframework.amqp.core.FanoutExchange
    **/
    @Bean
    public FanoutExchange exchange()
        return   new FanoutExchange(EXCHANGE_NAME,true,false);
    

    /**
     * @description: 声明绑定关系 注意：后⾯多了with携带routing-key
     * @method: binding
     * @author: wang fei
     * @date: 2023/2/4 17:17:04
     * @param: [queue, exchange]
     * @return: org.springframework.amqp.core.Binding
    **/
    @Bean
    public Binding binding(Queue queue, Exchange exchange) 
        return BindingBuilder.bind(queue).to(exchange).with("product.*").noargs();

编写消费者

package com.example.consumer;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

/**
 * @BelongsProject: RabbitMqLearn
 * @BelongsPackage: com.wang.config.consumer
 * @Author: wang fei
 * @CreateTime: 2023-02-04  17:20
 * @Description: TODO MQ 订阅者
 * @Version: 1.0
 */
@Component
public class MyConsumer 

    /**
     * @description: 监听队列，当队列中有消息的时候，该⽅法会被回调，⽤来消费消息 消费端的幂等性的实现
     * @method: receive
     * @author: wang fei
     * @date: 2023/2/4 17:22:35
     * @param: [message]
     * @return: void
     **/
    @RabbitListener(queues = "my_boot_potic_queue")
    public void receive(Message message, Channel channel) throws IOException 
        byte[] msg = message.getBody();
        //获得消息的业务id
        String messageId = message.getMessageProperties().getHeader("spring_returned_message_correlation");
        //设置分布式锁
        Boolean lock=false;
//         lock = redisTemplate.opsForValue().setIfAbsent(messageId, 1, 100000,TimeUnit.MILLISECONDS);
        //⼿动ack
        if (lock) 
            //做消费
            //⼿动ack
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
         else 
            //不做消费
            System.out.println("已重复消费");
            channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
        
        System.out.println("收到消息：" + new String(msg));

2 ）编写⽣产者

调整⽣产者的配置

package com.example.config;

import org.springframework.amqp.core.FanoutExchange;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;



/**
 * @BelongsProject: RabbitMqLearn
 * @BelongsPackage: com.wang.config
 * @Author: wang fei
 * @CreateTime: 2023-02-04  17:30
 * @Description: TODO
 * @Version: 1.0
 */
@Configuration
public class RabbitConfig  
    private static String EXCHANGE_NAME = "my_boot_topic_exchange";

    /**
     * 声明交换机
     */
    @Bean
    public FanoutExchange exchange()
        return new FanoutExchange(EXCHANGE_NAME,true,false);

发消息时携带routing-key

package com.example;

import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

@SpringBootTest
class MySpringBootTopicProducerApplicationTests 
    @Autowired
    RabbitTemplate rabbitTemplate;

    @Test
    void contextLoads() 
        String msg="Hello spring boot rabbit mq-topic";
        rabbitTemplate.convertAndSend("my_boot_topic_exchange","product.add", msg);
        System.out.println("发送信息成功");

5.⼿动ack的实现

在配置⽂件中添加⼿动ack的配置

server:
  port: 8090
spring:
  application:
    name: my-spring-boot-topic-consumer
  rabbitmq:
    port: 5672
    host: 192.168.206.130
    username: wang
    password: 123456
    virtual-host:  /test1
      # 关闭⾃动ack，设置成⼿动ack
    listener:
      simple:
       acknowledge-mode: manual

在消费者中进⾏⼿动ack

七、消息的可靠性投递

1.通过confirm机制保证⽣产者消息能够投递到MQ

在spring项⽬中做confirm

package com.wang.confirm;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.wang.utils.RabbitMqUtil;

import java.io.IOException;


/**
 * @BelongsProject: RabbitMqLearn
 * @BelongsPackage: com.wang
 * @Author: wang fei
 * @CreateTime: 2023-02-03  15:58
 * @Description: TODO rabbitmq-topics模式 生产者  通过confirm机制保证⽣产者消息能够投递到MQ
 * @Version: 1.0
 */
public class MyProducer 
    //定义交换机名称
    public  static String EXCHANGE_NAME = "my_topic_exchange";

    public static void main(String[] args) throws Exception 
        Connection connection=RabbitMqUtil.getConnection();
        Channel channel = connection.createChannel();
        //声名交换机
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");
        //开启confirm机制
        channel.confirmSelect();
        //设置confrim的监听器
        channel.addConfirmListener(new ConfirmListener() 
            //当消息发送成功将会执⾏这⾥的⽅法
            @Override
            public void handleAck(long l, boolean b) throws IOException 
                System.out.println("消息已经成功投递");
            
            //当消息发送失败会执⾏这⾥的⽅法，通过重试机制，进⾏重新投递，如果重新投递的次数达到阈值，那么就需要⼈⼯介⼊
            @Override
            public void handleNack(long l, boolean b) throws IOException 
                System.out.println("消息投递失败");
            
        );

        //发送信息
        String message = "topics模式!";
        channel.basicPublish(EXCHANGE_NAME, "product.add.one", null, message.getBytes());
        System.out.println("消息发送成功");
        //断开链接
        channel.close();
        connection.close();

在springboot中实现

步骤⼀：修改⽣产者的配置

server:
  port: 8090
spring:
  application:
    name: my-spring-boot-topic-producer
  rabbitmq:
    port: 5672
    host: 192.168.206.130
    username: wang
    password: 123456
    virtual-host:  /test1
    # 开启confirm simple:简单的执⾏ack的判断；correlated: 执⾏ack的时候还会携带数据;none: 不ack 默认的
    publisher-confirm-type: correlated
      # 开启return机制
    publisher-returns: true

publisher-confirm-type: 有三种配置：

simple:简单的执⾏ack的判断；在发布消息成功后使⽤rabbitTemplate调⽤
waitForConfirms或waitForConfirmsOrDie⽅法等待broker节点返回发送结果，根据返回结果来判断下⼀步的逻辑。但是要注意的是当waitForConfirmsOrDie⽅法如果返回false则会关闭channel。
correlated: 执⾏ack的时候还会携带数据（消息的元数据）;
none: 禁⽤发布确认模式，默认的

步骤⼆：编写⼀个 ConfirmCallback 的实现类（监听器），并注⼊到 R

package com.example.confirm_retrun;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.Objects;

/**
 * @BelongsProject: RabbitMqLearn
 * @BelongsPackage: com.wang.config
 * @Author: wang fei
 * @CreateTime: 2023-02-04  17:30
 * @Description: TODO
 * @Version: 1.0
 */
@Component
public class RabbitConfirmConfig implements RabbitTemplate.ConfirmCallback
    @Autowired
    private RabbitTemplate rabbitTemplate;


    /**
     * @description: 注⼊当前监听器到RabbitTemplate
     * @method: init
     * @author: wang fei
     * @date: 2023/2/6 17:45:56
     **/
    @PostConstruct
    public void init() 
        rabbitTemplate.setConfirmCallback(this);
    

    @Override
    public void confirm(CorrelationData correlationData, boolean b, String s) 
        String dataId = "";
        if (Objects.nonNull(correlationData)) 
            dataId = correlationData.getId();
        
        if (b) 
            //信息发送成功
            System.out.printf("消息发送成功:" + dataId);
         else 
            //信息发送失败
            System.out.printf("消息发送失败:" + s);

2.通过return机制保证消息在rabbitmq中能够成功的投递到队列⾥

⽣产者将消息投递到 mq 的交换机上 ——Confirm 机制来保证的。如果交换机没办法将消息投递到队列上，就可以通过Return 机制来进⾏重试。 1) 在 spring 项⽬中

package com.wang.returns;

import com.rabbitmq.client.*;

import com.wang.utils.RabbitMqUtil;

import java.io.IOException;


/**
 * @BelongsProject: RabbitMqLearn
 * @BelongsPackage: com.wang
 * @Author: wang fei
 * @CreateTime: 2023-02-03  15:58
 * @Description: TODO rabbitmq-topics模式 生产者  通过confirm机制保证⽣产者消息能够投递到MQ  通过return机制保证消息在rabbitmq中能够 成功的投递到队列⾥
 * @Version: 1.0
 */
public class MyProducer 
    //定义交换机名称
    public  static String EXCHANGE_NAME = "my_topic_exchange";

    public static void main(String[] args) throws Exception 
        Connection connection=RabbitMqUtil.getConnection();
        Channel channel = connection.createChannel();
        //声名交换机
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");
        //开启confirm机制
        channel.confirmSelect();
        //设置confrim的监听器
        channel.addConfirmListener(new ConfirmListener() 
            //当消息发送成功将会执⾏这⾥的⽅法
            @Override
            public void handleAck(long l, boolean b) throws IOException 
                System.out.println("消息已经成功投递");
            
            //当消息发送失败会执⾏这⾥的⽅法，通过重试机制，进⾏重新投递，如果重新投递的次数达到阈值，那么就需要⼈⼯介⼊
            @Override
            public void handleNack(long l, boolean b) throws IOException 
                System.out.println("消息投递失败");
            
        );

        // 开启return机制
        channel.addReturnListener(new ReturnListener() 
            @Override
            public void handleReturn(int replyCode, String
                    replyText, String exchange, String routingKey,
                                     AMQP.BasicProperties properties, byte[] body) throws IOException 
                //如果消息没有成功抵达队列，此⽅法将会被调⽤
                System.out.println("消息没有抵达队列");
            
        );

        //发送信息
        String message = "topics模式!";
        channel.basicPublish(EXCHANGE_NAME, "product.add.one", true,null, message.getBytes());
        System.out.println("消息发送成rabbitmq学习笔记(代码片段)
文章目录RabbitMq(消息中间件)1.概念：是基于队列模式实现的异步/同步的传输数据。2.传统的Http请求存在哪些缺点？3.mq的使用场景？4.mq服务器如何保证消息不丢失？5.VirtualHost?Exchange分派我们的消息存放在哪个队列...  查看详情  
                
学习笔记《rabbitmq实战指南》笔记(代码片段)
本文摘录总结自《RabbitMQ实战指南》。一、消息中间件消息队列中间件（MessageQueueMiddleware，简称为MQ）是指利用高效可靠的消息传递机制进行与平台无关的数据交流，并基于数据通信来进行分布式系统的集成。它...  查看详情  
                
学习笔记《rabbitmq实战指南》笔记(代码片段)
本文摘录总结自《RabbitMQ实战指南》。一、消息中间件消息队列中间件（MessageQueueMiddleware，简称为MQ）是指利用高效可靠的消息传递机制进行与平台无关的数据交流，并基于数据通信来进行分布式系统的集成。它...  查看详情  
                
rabbitmq学习笔记(代码片段)
视频教程【编程不良人】MQ消息中间件之RabbitMQ以及整合SpringBoot2.x实战教程1.MQ引言1.1什么是MQMQ(MessageQuene)：消息队列，也叫消息中间件。通过典型的生产者和消费者模型，生产者不断向消息队列中生产消息，消费...  查看详情  
                
rabbitmq学习笔记-p2（springamqp）(代码片段)
...超级推荐！！SpringAMQP1.初识SpringAMQPSpringAMQP是基于RabbitMQ封装的一套模板，并且还利用SpringBoot对其实现了自动装配，使用起来非常方便。SpringAmqp的官方地址：https://spring.io/projects/spring-am  查看详情  
                
rabbitmq学习笔记(代码片段)
rabbitmqAMQP协议安装如何使用AMQP协议生产者消费者队列一个队列存多条消息消息连接发消息和接收消息都是必须先建立连接通道（channel）一个连接有多个通道，消息会在通道里面流转，之后到达消费者交换机（...  查看详情  
                
rabbitmq学习笔记六:话题（topics）(代码片段)
前一篇我们已经改善了我们的日志系统，不用fanout类型的转发器愚蠢的广播消息，而是使用direct类型的转发器，能够选择性的接收我们想要接收的消息。参见Rabbit学习笔记五：路由选择（Routing）http://blog.cs...  查看详情  
                
rabbitmq学习笔记四：消息publish/subscrible(代码片段)
...任务意外被杀死，会转发给另外的工作者，参见[Rabbitmq学习笔记三：工作队列workqueue](http://blog.csdn.net/u010416588/article/details/54629334)。一发布、订阅简介  查看详情  
                
分布式服务框架学习笔记10rabbitmq在centos下的安装(代码片段)
分布式服务框架学习笔记10RabbitMQ在CentOS下的安装一、准备二、安装erlang1.安装依赖3.RabbitMQ默认端口号自行启用三、安装RabbitMQ三、使用WEB管理界面1.启用web2.用户管理命令一、准备centos8要先安装erlang二、安装erlang1.安装依赖yum-yins...  查看详情  
                
分布式消息中间件之rabbitmq学习笔记[一](代码片段)
...，陆续的整理一些中间件的笔记今天和小伙伴们分享RabbitMQ相关笔记博文偏理论,内容涉及：RabbitMQ的简单介绍AMQP协议标准介绍RabbitMQDemo食用方式：了解生产者消费者模式通过本文，对RabbitMQ有大概认识户外依然大...  查看详情  
                
分布式消息中间件之rabbitmq学习笔记[一](代码片段)
...，陆续的整理一些中间件的笔记今天和小伙伴们分享RabbitMQ相关笔记博文偏理论,内容涉及：RabbitMQ的简单介绍AMQP协议标准介绍RabbitMQDemo食用方式：了解生产者消费者模式通过本文，对RabbitMQ有大概认识户外依然大...  查看详情  
                
springcloudstream消息驱动设计思想以及整合rabbitmq消息队列案例--学习笔记(代码片段)
一，RabbitMQ的安装和配置并启动参见RabbitMQ的安装和配置https://blog.csdn.net/weixin_43025151/article/details/123186641RabbitMQ启动成功：二，SpringCloudStream消息驱动1，设计思想一个标准的消息队列MQ,如下图：为什么用Cloud  查看详情  
                
rabbitmq超详细学习笔记（章节清晰+通俗易懂）(代码片段)
...学习，最近花了一段时间系统学习了当下最为主流的RabbitMQ消息队列，学习过程中也随时记录，刚开始学习的时候懵懵懂懂，做的笔记都比较杂乱，系统学习完后我将笔记内容不断反复修改，对章节进行设...  查看详情  
                
rabbitmq学习笔记(代码片段)
RabbitMq学习笔记1.消息队列1.1MQ相关概念1.1.1什么是MQ？1.1.2为什么要用MQ？1.2.3MQ的分类1.2.4MQ的选择1.2.5为什么选择RabbitMQ？1.2.5MQ对比分析图在这里插入图片描述1.2RabbitMQ1.2.1.RabbitMQ的概念1.2.2.四大核心概念1.2.3RabbitMQ6大模...  查看详情  
                
rabbitmq学习笔记(代码片段)
目录一、引⾔二、RabbitMQ介绍三、RabbitMQ安装四、RabbitMQ架构1.官⽅的简单架构图2.RabbitMQ的完整架构图3.查看图形化界⾯并创建⼀个VirtualHost五、RabbitMQ的队列模式1.RabbitMQ的通讯⽅式2.HelloWorld模式-简单队列模式3.work队列模式：...  查看详情  
                
rabbitmq学习笔记-p1（初识mq&快速入门）(代码片段)
...频指路👉B站黑马微服务超级推荐！！！RabbitMQ1.初识MQ1.1同步和异步通讯微服务间通讯有同步和异步两种方式：同步通讯：就像打电话，需要实时响应异步通讯：就像发邮件，不需要马上回复两...  查看详情  
                
rabbitmq学习笔记（持续更新ing）(代码片段)
...消息模式发布订阅模型快速入门（java）首先安装rabbitmq（单机版）rabbitmq的安装（官网文档）在我自己租的云服务器上，直接用docker进行安装（一行命令搞定）dockerrun-it--rm--namerabbitmq-p5672:5672-p156...  查看详情  
                
rabbitmq学习笔记五：rabbitmq之优先级消息队列
RabbitMQ优先级队列注意点：1、只有当消费者不足，不能及时进行消费的情况下，优先级队列才会生效2、RabbitMQ3.5以后才支持优先级队列代码在博客：RabbitMQ学习笔记三：Java实现RabbitMQ之与Spring集成 最后面有下载地址，只是做...  查看详情

正文

rabbitmq学习笔记(代码片段)

一、引⾔

二、RabbitMQ介绍

三、RabbitMQ安装

四、RabbitMQ架构

1.官⽅的简单架构图

2.RabbitMQ的完整架构图

3.查看图形化界⾯并创建⼀个Virtual Host

五、RabbitMQ的队列模式

1.RabbitMQ的通讯⽅式

2.HelloWorld模式-简单队列模式

3. work 队列模式： 能者多劳模式

4. 发布订阅模式-fanout

5.routing模式-direct

6.topics模式

六、在Springboot中使⽤RabbitMQ

1.使⽤发布订阅模式

2.使⽤topic模式

5.⼿动ack的实现

七、消息的可靠性投递

1.通过confirm机制保证⽣产者消息能够投递到MQ

2.通过return机制保证消息在rabbitmq中能够成功的投递到队列⾥

rabbitmq学习笔记(代码片段)

学习笔记《rabbitmq实战指南》笔记(代码片段)

学习笔记《rabbitmq实战指南》笔记(代码片段)

rabbitmq学习笔记(代码片段)

rabbitmq学习笔记-p2（springamqp）(代码片段)

rabbitmq学习笔记(代码片段)

rabbitmq学习笔记六:话题（topics）(代码片段)

rabbitmq学习笔记四：消息publish/subscrible(代码片段)

分布式服务框架学习笔记10rabbitmq在centos下的安装(代码片段)

分布式消息中间件之rabbitmq学习笔记[一](代码片段)

分布式消息中间件之rabbitmq学习笔记[一](代码片段)

springcloudstream消息驱动设计思想以及整合rabbitmq消息队列案例--学习笔记(代码片段)

rabbitmq超详细学习笔记（章节清晰+通俗易懂）(代码片段)

rabbitmq学习笔记(代码片段)

rabbitmq学习笔记(代码片段)

rabbitmq学习笔记-p1（初识mq&快速入门）(代码片段)

rabbitmq学习笔记（持续更新ing）(代码片段)

rabbitmq学习笔记五：rabbitmq之优先级消息队列

3. work 队列模式：能者多劳模式