RabbitMQ学习笔记
Posted Traveler飞
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了RabbitMQ学习笔记相关的知识，希望对你有一定的参考价值。
目录
2.通过return机制保证消息在rabbitmq中能够成功的投递到队列⾥
一、引⾔
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安装
环境：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();
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->);
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);
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();
- 创建队列
- 创建交换机
- 把队列绑定在交换机上
- 让消费者监听队列
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
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();
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 ok1 ）编写⽣产者
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();
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>server:
port: 8090
spring:
application:
name: my-spring-boot-producer
rabbitmq:
port: 5672
host: 192.168.206.130
username: wang
password: 123456
virtual-host: /test11.使⽤发布订阅模式
- 编写配置类
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);
- 编写消费消息的⽅法
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));
- 编写配置类
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));
- 调整⽣产者的配置
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- simple:简单的执⾏ack的判断；在发布消息成功后使⽤rabbitTemplate调⽤
- waitForConfirms或waitForConfirmsOrDie⽅法等待broker节点返回发送结果，根 据返回结果来判断下⼀步的逻辑。但是要注意的是当waitForConfirmsOrDie⽅法如果返回false则会关闭channel。
- correlated: 执⾏ack的时候还会携带数据（消息的元数据）;
- none: 禁⽤发布确认模式， 默认的
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学习笔记的主要内容，如果未能解决你的问题，请参考以下文章