csdn推荐
RabbitMq
官网:RabbitMQ: One broker to queue them all | RabbitMQ
什么是MQ？
mq就是消息队列，消息队列遵循这先入先出原则。一般用来解决应用解耦，异步消息，流量削峰等问题，实现高性能，高可用，可伸缩和最终一致性架构。
rabbitMq的四大核心
RabbitMq的安装
RabbitMQ是一个开源的遵循 AMQP协议实现的基于 Erlang语言编写，即需要先安装部署Erlang环境再安装RabbitMQ环境。
查看兼容关系：Erlang Version Requirements | RabbitMQ
百度云地址：
链接：百度网盘 请输入提取码 提取码：6666
本篇文章使用版本：3.8.8，liunx7-cenOs7
#在存放位置执行以下指令
rpm -ivh erlang-21.3-1.el7.x86_64.rpm
#安装socat
yum install socat -y
#安装mq
rpm -ivh rabbitmq-server-3.8.8-1.el7.noarch.rpm 启动
#开机自动启动
chkconfig rabbitmq-server on
#启动服务
/sbin/service rabbitmq-serve start
#查看启动
/sbin/service rabbitmq-serve status
#停止服务
/sbin/service rabbitmq-serve stop坑：执行以上指令无效，重新执行下面指令
systemctl start rabbitmq-server.service #启动
systemctl status rabbitmq-server.service#查看状态安装可视化界面
#尽量停止服务，在安装
#安装可视化界面
rabbitmq-plugins enable rabbitmq_management访问地址：:15672/
如果访问不了，查看防火墙是够关闭 systemctl stop firewalld 关闭防火墙，访问成功后走rabbitmq的基本指令
卸载MQ：
systemctl stop rabbitmq-server
yum list | grep rabbitmq
yum -y remove rabbitmq-server.noarch
yum list | grep erlang
yum -y remove erlang-*
rm -rf /usr/lib64/erlang
rm -rf /var/lib/rabbitmq
rm -rf /usr/local/erlang
rm -rf /usr/local/rabbitmqdocker安装
docker pull rabbitmq:3-management#运行
docker run
-e RABBITMQ_DEFAULT_USER=itcast
-e RABBITMQ_DEFAULT_PASS=123321
--name mq
--hostname mq1
-p 15672:15672 #网页访问端口
-p 5672:5672 #mq连接端口
-d
rabbitmq:3-managementrabbitMq基本指令
#查看用户
rabbitmqctl list_users
#添加用户
rabbitmqctl add_user admin 123456
#设置角色 (超级管理员)
rabbitmqctl set_user_tags admin administrator
#设置权限
rabbitmqctl set_permissions -p '/' admin '.*' '.*' '.*'登录后也可以在此界面添加用户
对接java（入门）
创建一个maven工程：
4.0.0
com.chen
mq
1.0-SNAPSHOT
5.8.0
2.6
com.rabbitmq
amqp-client
${rabbitmq.version}
commons-io
commons-io
${common.version}
org.apache.maven.plugins
maven-compiler-plugin
8
8
生产者：
package com.chen.rabbitmq.one;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
/**
* 生产者
*/
public class production {
private static final String MQ_KEY="holle";
public static void main(String[] args)throws Exception {
// 创建rabbitmq的工厂
ConnectionFactory factory = new ConnectionFactory();
// 连接地址ip
factory.setHost("172.17.18.162");
// 用户名
factory.setUsername("admin");
// 密码
factory.setPassword("123456");
// 创建连接
Connection connection = factory.newConnection();
// 创建通道
Channel channel = connection.createChannel();
// 生产队列
// 参数一：队列名称
// 参数二：持久性（默认为false）
// 参数三：该队列是否可以有多个消费者，是否消息共享
// 参数四：是否自动删除
// 参数五：其他参数
channel.queueDeclare(MQ_KEY,true,false,false,null);
/**
* 发送一个消费者
* 1.发送到那个交换机
* 2.路由的key值是哪个 本次是队列的名称
* 3.其他参数
* 4.发送消息的消息体
*/
channel.basicPublish("",MQ_KEY,null,"holle word".getBytes());
System.out.println("消息发送成功！");
}
}测试是否发送成功：
消费者：
package com.chen.rabbitmq.one;
import com.rabbitmq.client.*;
import java.io.IOException;
public class Consumption {
private static final String MQ_KEY="holle";
public static void main(String[] args) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("172.17.18.162");
factory.setUsername("admin");
factory.setPassword("123456");
// 创建一个新的连接
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
/*
参数：
1: 消费哪个队列
2.消费成功之后是否要自动应答， true 带边自动应答 false 手动
3.消费者未成功的回调
4.消费者取录成功的回调
*/
channel.basicConsume(MQ_KEY, true,(DeliverCallback) (consumerTag, message) -> System.out.println(new String(message.getBody())),
(CancelCallback) (consumerTag)-> System.out.println(consumerTag));
}
}工作队列：
工作队列(又称任务队列)的主要思想是避免立即执行资源密集型任务，而不得不等待它完成。相反我们安排任务在之后执行。我们把任务封装为消息并将其发送到队列。在后台运行的工作进程将弹出任务并最终执行作业。当有多个工作线程时，这些工作线程将一起处理这些任务。
1.线程轮询
类似nginx的负载均衡(轮询)，线1一次，线2一次。
工具类：
package com.chen.rabbitmq.tow.utils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class RabbitUtils {
public static Channel rabbitConnection() throws Exception{
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("172.17.18.162");
factory.setUsername("admin");
factory.setPassword("123456");
// 创建一个新的连接
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
return channel;
}
}
生产者：
package com.chen.rabbitmq.tow.test;
import com.chen.rabbitmq.tow.utils.RabbitUtils;
import com.rabbitmq.client.Channel;
import java.util.Scanner;
public class Production {
private final static String MQ_KEY="word";
// 生产者
public static void production() throws Exception{
Channel channel = RabbitUtils.rabbitConnection();
Scanner scanner = new Scanner(System.in);
//生产队列
channel.queueDeclare(MQ_KEY,true,false,false,null);
while (scanner.hasNext()){
String next = scanner.next();
channel.basicPublish("",MQ_KEY,null,next.getBytes());
System.out.println("消息发布成功-> "+next);
}
}
public static void main(String[] args) throws Exception{
production();
}
}
消费者：
package com.chen.rabbitmq.tow.test;
import com.chen.rabbitmq.tow.utils.RabbitUtils;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
public class Consumption {
private final static String MQ_KEY="word";
// 消费者
public static void consumption() throws Exception{
// 获取连接队列
Channel channel = RabbitUtils.rabbitConnection();
channel.basicConsume(MQ_KEY,true,(DeliverCallback)(consumerTag,message)->{
System.out.println(new String(message.getBody()));
},(CancelCallback)(tag)->{
System.out.println(tag);
System.out.println("中断了");
});
}
public static void main(String[] args) throws Exception{
consumption();
}
}idea开启两个线程。
消息应答 1.自动应答
RabbitMQ 是一个广泛使用的开源消息代理，它支持多种消息协议，例如 AMQP、MQTT、STOMP 等。在 RabbitMQ 中，自动应答（Automatic Acknowledgement，Auto-ack）是一种消息确认机制，用于标记消息是否已被成功接收和处理。了解自动应答的概念，对于构建可靠、高效的消息传递系统非常重要。
当消费者接收并处理来自 RabbitMQ 的消息时，通常会使用消息确认（acknowledgements）机制来告知 RabbitMQ 该消息已经成功处理。这样一来，RabbitMQ 就可以确保消息不会意外丢失。然而，这种确认过程可能会导致一定的延迟和额外开销。为了解决这个问题，RabbitMQ 提供了自动应答机制。
在自动应答模式下，消费者接收到消息后，RabbitMQ 会立即将该消息标记为已处理。这意味着消费者不需要显式地发送确认（ack）消息给 RabbitMQ。这种机制可以降低延迟，提高消息传递的速度，但是也存在一定的风险。因为消息一旦被发送出去，RabbitMQ 就认为它已经成功处理，而实际上消费者可能还没有完成对消息的处理。如果消费者在处理消息时发生故障，那么这个消息可能会丢失。
2.手动应答
方法：
Channel.basicAck (用于肯定确认)
RabbitMQ已知道该消息并且成功的处理消息，可以将其丢弃了
Channel.basicNack(用于否定确认)
Channel.basicReject (用于否定确认)
与 Channel.basicNack 相比少一个参数 不处理该消息了直接拒绝，可以将其丢弃了
Multiple
//源码
public void basicAck(long deliveryTag, boolean multiple) throws IOException {
this.delegate.basicAck(deliveryTag, multiple);
}multiple 的 true 和 false 代表不同意思:
true 代表批量应答 channel 上未应答的消息
比如说 channel 上有传送 tag 的消息 5,6,7,8 当前 tag 是 8 那么此时5-8 的这些还未应答的消息都会被确认收到消息应答
2.false 只会应答 tag=8 的消息 5,6,7 这三个消息依然不会被确认收到消息应答
消息重新入队
为了解决消息丢失问题。
具体代码：
生产者：
package com.chen.rabbitmq.three;
import com.chen.rabbitmq.tow.utils.RabbitUtils;
import com.rabbitmq.client.Channel;
import java.nio.charset.StandardCharsets;
import java.util.Scanner;
public class Pro {
private static final String MQ_KEY="mqkey";
public static void pro() throws Exception{
Channel channel = RabbitUtils.rabbitConnection();
channel.queueDeclare(MQ_KEY,true,false,false,null);
Scanner scanner = new Scanner(System.in);
while (scanner.hasNext()){
String next = scanner.next();
channel.basicPublish("",MQ_KEY,null,scanner.next().getBytes());
System.out.println("消息发布成功-> "+next);
}
}
public static void main(String[] args) throws Exception {
pro();
}
}消费者1：
package com.chen.rabbitmq.three;
import com.chen.rabbitmq.tow.utils.RabbitUtils;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
//消费者1
public class Word1 {
public static final String MQ_KEY="mqkey";
public static void word() throws Exception{
Channel channel = RabbitUtils.rabbitConnection();
channel.basicConsume(MQ_KEY,false,(DeliverCallback) (consumerTag,message)->{
// 睡眠1s
try {
Thread.sleep(1*1000);
System.out.println("Word1接收到消息->"+new String(message.getBody()));
// 参数一：tag标记 参数二：是否批量
channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
} catch (InterruptedException e) {
e.printStackTrace();
}
},(CancelCallback) e->{
System.out.println("消息中断"+e);
} );
}
public static void main(String[] args) throws Exception {
word();
}
}消费者2：
package com.chen.rabbitmq.three;
import com.chen.rabbitmq.tow.utils.RabbitUtils;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
//消费者1
public class Word2 {
public static final String MQ_KEY="mqkey";
public static void word() throws Exception{
Channel channel = RabbitUtils.rabbitConnection();
channel.basicConsume(MQ_KEY,false,(DeliverCallback) (consumerTag,message)->{
// 睡眠10s
try {
Thread.sleep(10*1000);
System.out.println("Word2接收到消息->"+new String(message.getBody()));
// 参数一：tag标记 参数二：是否批量
channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
} catch (InterruptedException e) {
e.printStackTrace();
}
},(CancelCallback) e->{
System.out.println("消息中断"+e);
} );
}
public static void main(String[] args) throws Exception{
word();
}
}经测试会发现，消费者1为第一个接收到消息，接下来当生产者在生产出一条消息，应到消费者2接收到消息，但是此时消费者2突然出现宕机，使用了应答机制，消息则会重新打到消费者1；
持久化设置 1.队列持久化
作用:当rabbitmq宕机后，重启队列依然存在
//创建队列时的第二个参数为设置持久化
channel.queueDeclare(MQ_KEY,true,false,false,null);2.消息持久化
作用：当rabbitmq宕机了重新启动，发送的消息依然存在。
下面的方法不是绝对的能保证消息的持久化
//生产者
private static final String MQ_KEY="mqkey";
public static void pro() throws Exception{
Channel channel = RabbitUtils.rabbitConnection();
channel.queueDeclare(MQ_KEY,true,false,false,null);
Scanner scanner = new Scanner(System.in);
while (scanner.hasNext()){
String next = scanner.next();
//MessageProperties.PERSISTENT_TEXT_PLAIN 消息持久化
channel.basicPublish("",MQ_KEY,MessageProperties.PERSISTENT_TEXT_PLAIN,scanner.next().getBytes());
System.out.println("消息发布成功-> "+next);
}
}3.发布确认
完成以上两步还不足以持久化，要把发布确认加上。
//默认是不开启的
Channel channel = RabbitUtils.rabbitConnection();
channel.confirmSelect();//开启发布确认发布确认的策略:
1.单个确认发布
这个发布确认是同步的，需等待确认一次在发布下一次，一手交钱一手交货原则
缺点：发布速度特别慢
//单个确认
public static void one() throws Exception{
Channel channel = RabbitUtils.rabbitConnection();
//开启发布确认
channel.confirmSelect();
String uuid = UUID.randomUUID().toString();
//创建队列
channel.queueDeclare(uuid,true,false,false,null);
//开始时间
long begin = System.currentTimeMillis();
for (Integer i = 0; i < COUNT; i++) {
String message = i + "";
channel.basicPublish("",uuid,null,message.getBytes());
//发布确认
boolean flag = channel.waitForConfirms();
if(flag){
System.out.println("消息确认成功！");
}
}
long last = System.currentTimeMillis();
System.out.println("耗时："+(last-begin));
}2.批量确认发布
发布速度相对单个发布确认要快，但是当其中一条消息出现异常，将无法查找到那个消息丢失 。
//批量
public static void batch() throws Exception{
Channel channel = RabbitUtils.rabbitConnection();
String uuid = UUID.randomUUID().toString();
//开启消息确认
channel.confirmSelect();
//创建队列
channel.queueDeclare(uuid,true,false,false,null);
//这个这个变量用记录发布值
Integer messageCount=100;
Integer record =0;
//开始时间
long begin = System.currentTimeMillis();
for (Integer i = 0; i < COUNT; i++) {
record++;
String message=i+"";
//发布消息
channel.basicPublish("",uuid,null,message.getBytes());
if(messageCount.equals(record)){
channel.waitForConfirms();
record=0;
}
}
long last = System.currentTimeMillis();
System.out.println("耗时"+(last-begin));
}3.异步确认发布(推荐使用)
异步确认虽然比上的两个代码复杂，但同时也解决了上面两种方式遗留下来的问题。
public static void asyn() throws Exception{
Channel channel = RabbitUtils.rabbitConnection();
//开启发布确认
channel.confirmSelect();
String uuid = UUID.randomUUID().toString();
//创建队列
channel.queueDeclare(uuid,true,false,false,null);
//开始时间
long begin = System.currentTimeMillis();
// 创建一个线程的ListMap用于记录 ----》处理异步未确认的消息
ConcurrentSkipListMap map = new ConcurrentSkipListMap();
// 监听消息
channel.addConfirmListener((deliveryTag, multiple)->{
if(multiple){
ConcurrentNavigableMap longStringConcurrentNavigableMap =
map.headMap(deliveryTag);
longStringConcurrentNavigableMap.clear();
}else{
map.remove(deliveryTag);
}
System.out.println("确认消息："+deliveryTag);
},(deliveryTag, multiple)->{
String message = map.get(deliveryTag);
System.out.println("发送失败的数据是："+message+"未确认消息："+deliveryTag+"-----失败");
});
for (Integer i = 0; i < COUNT; i++) {
String message=""+i;
channel.basicPublish("",uuid,null,message.getBytes());
//获取信道的标识，存入消息
map.put(channel.getNextPublishSeqNo(),message);
}
long last = System.currentTimeMillis();
System.out.println("耗时："+(last-begin));
}不公平分发原则(能者多劳原则)
在上面中的所有例子都是尊寻这轮询的规则去执行的，问题:当其中的一台服务响应特别慢时就会影响到整体的效率。
channel.basicQos(1);
//消费者
public static void word() throws Exception{
Channel channel = RabbitUtils.rabbitConnection();
//设置不公平分发
channel.basicQos(1);
channel.basicConsume(MQ_KEY,false,(DeliverCallback) (consumerTag,message)->{
try {
//模拟虚拟机延迟
Thread.sleep(1*1000);
System.out.println("Word2接收到消息->"+new String(message.getBody()));
channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
} catch (InterruptedException e) {
e.printStackTrace();
}
},(CancelCallback) e->{
System.out.println("消息中断"+e);
} );
}也可以用来设置预期值！
//消费者1
public static void word2() throws Exception{
Channel channel = RabbitUtils.rabbitConnection();
//设置预期值
channel.basicQos(3);
channel.basicConsume(MQ_KEY,false,(DeliverCallback) (consumerTag,message)->{
try {
//模拟虚拟机延迟
Thread.sleep(1*1000);
System.out.println("Word2接收到消息->"+new String(message.getBody()));
channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
} catch (InterruptedException e) {
e.printStackTrace();
}
},(CancelCallback) e->{
System.out.println("消息中断"+e);
} );
}
//消费者2
public static void word2() throws Exception{
Channel channel = RabbitUtils.rabbitConnection();
//设置预期值
channel.basicQos(5);
channel.basicConsume(MQ_KEY,false,(DeliverCallback) (consumerTag,message)->{
try {
//模拟虚拟机延迟
Thread.sleep(10*1000);
System.out.println("Word2接收到消息->"+new String(message.getBody()));
channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
} catch (InterruptedException e) {
e.printStackTrace();
}
},(CancelCallback) e->{
System.out.println("消息中断"+e);
} );
}交换机
在RabbitMQ中，生产者发送消息不会直接将消息投递到队列中，而是先将消息投递到交换机中， 在由交换机转发到具体的队列， 队列再将消息以推送或者拉取方式给消费者进行消费
绑定(bindings)
与交换机产生关系，并且能有routekey控制发送消息给哪个队列。
fanout交换机(扇形)
扇形交换机是最基本的交换机类型，它所能做的事清非常简单广播消息。扇形交换机会把能接收到的消息全部发送给绑定在自己身上的队列。因为广播不需要＇思考”，所以扇形交换机处理消息的速度也是所有的交换机类型里面最快的。
//消费者
public class Word {
// 交换机名称
private static String EXCHANGE_NAME="logs";
public static void main(String[] args) throws Exception {
Channel channel = RabbitUtils.rabbitConnection();
// 声明一个交换机
channel.exchangeDeclare(EXCHANGE_NAME,"fanout");
// 声明一个队列 临时队列
String queue = channel.queueDeclare().getQueue();
// 绑定交换机与队列
channel.queueBind(queue,EXCHANGE_NAME,"");
System.out.println("等待消息~");
//消费者取消消息时回调接口
channel.basicConsume(queue,true, (consumerTag,message)->{
System.out.println("word1控制台打印接收消息:"+new String(message.getBody(),"UTF-8"));
},cancelCallback->{});
}
}
public class Word2 {
// 交换机名称
private static String EXCHANGE_NAME="logs";
public static void main(String[] args) throws Exception {
Channel channel = RabbitUtils.rabbitConnection();
// 声明一个交换机
channel.exchangeDeclare(EXCHANGE_NAME,"fanout");
// 声明一个队列 临时队列
String queue = channel.queueDeclare().getQueue();
// 绑定交换机与队列
channel.queueBind(queue,EXCHANGE_NAME,"");
System.out.println("等待消息~");
//消费者取消消息时回调接口
channel.basicConsume(queue,true, (consumerTag,message)->{
System.out.println("word2控制台打印接收消息:"+new String(message.getBody(),"UTF-8"));
},cancelCallback->{});
}
}
//生产者
public class send {
// 交换机名称
private static String EXCHANGE_NAME="logs";
public static void main(String[] args) throws Exception {
Channel channel = RabbitUtils.rabbitConnection();
Scanner scanner = new Scanner(System.in);
while (scanner.hasNext()){
String next = scanner.next();
channel.basicPublish(EXCHANGE_NAME,"",null,next.getBytes("UTF-8"));
System.out.println("生产者发送消息："+next);
}
}
}直连交换机： Direct exchange
直连交换机的路由算法非常简单： 将消息推送到binding key与该消息的routing key相同的队列。
代码几乎类型fanout交换机，只需要指定routerkey即可。
主题交换机： Topic exchange
发送到主题交换机的 消息不能有任意的 routing key, 必须是由点号分开的一串单词，这些单词可以是任意的，但通常是与消息相关的一些特征。
如以下是几个有效的routing key:
"stock.usd.nyse", "nyse.vmw", "quick.orange.rabb 代", routing key的单词可以 有很多，最大限制是255 bytes。
Topic 交换机的 逻辑与 direct 交换机有点 相似 使用特定路由键发送的消息 将被发送到所有使用匹配绑定键绑定的队列 ，然而 ，绑定键有两个特殊的情况：
*表示匹配任意一个单词
#表示匹配任意—个或多个单词
比如上图：
发送routerkey为：ws.orange.rabbit那么对应的就是Q1，Q2
发送routerkey为：lazy.orange.elephant那么对应的就是Q1，Q2
//消费者
public class word1 {
private static final String EXCHANGE_NAME="topic_logs";
private static final String QUEUE_NAME="Q1";
public static void main(String[] args) throws Exception {
Channel channel = RabbitUtils.rabbitConnection();
// 创建交换机
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
// 创建队列
channel.queueDeclare(QUEUE_NAME,true,true,false,null);
// 绑定队列
channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"*.orange.*");
// 接收消息
channel.basicConsume(QUEUE_NAME,true,(consumerTag,message)->{
System.out.println("接收到的消息："+new String(message.getBody()));
},cancelCallback->{});
System.out.println("等下消息~");
}
}
public class word2 {
private static final String EXCHANGE_NAME="topic_logs";
private static final String QUEUE_NAME="Q2";
public static void main(String[] args) throws Exception {
Channel channel = RabbitUtils.rabbitConnection();
// 创建交换机
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
// 创建队列
channel.queueDeclare(QUEUE_NAME,true,true,false,null);
// 绑定队列
channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"*.*.rabbit");
channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"lazy.#");
// 接收消息
channel.basicConsume(QUEUE_NAME,true,(consumerTag,message)->{
System.out.println("接收到的消息："+new String(message.getBody()));
},cancelCallback->{});
System.out.println("等下消息~");
}
}
//生产者
public class send {
private static final String EXCHANGE_NAME="topic_logs";
public static void main(String[] args) throws Exception{
Channel channel = RabbitUtils.rabbitConnection();
Scanner scanner = new Scanner(System.in);
while (true){
System.out.println("请输入routerkey：");
String key = scanner.next();
System.out.println("请输入消息内容：");
String message = scanner.next();
channel.basicPublish(EXCHANGE_NAME,key,null,message.getBytes());
}
}
}死信队列
顾名思义：无法被消费的消息，一般来说，producer将消息投递broker或者直接到queue里了，consumer(消费者)从queue取出消息进行消费，但某些时间由特定原因导致queue中的某些消息无法被消费，这样如果没有后续的处理，就变成了死信。
应用场景：为了确保订单业务的消息数据不丢失，需要使用到RabbitMQ的死信队列机制，当消息被消息时发生了异常，这是就将消息存到死信中，还比如说：用户商城下单成功，并且点击支付后在指定时间支付时自动失效。
消息TTL过期时间测试：
//生产者
public class send {
private static final String NORMAL_EXCHANGE="normal_exchange";
public static final String NORMAL_QUEUE="normal_queue";
public static void main(String[] args) throws Exception {
Channel channel = RabbitUtils.rabbitConnection();
// 设置死信时间
AMQP.BasicProperties basicProperties =
new AMQP.BasicProperties().builder()
.expiration("10000").build();
for (int i = 0; i < 11; i++) {
String msg="info"+i;
channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",basicProperties,msg.getBytes());
}
}
}
//消费者1
public class C1 {
private static final String NORMAL_EXCHANGE="normal_exchange";
private static final String DEAD_EXCHANGE="dead_exchange";
public static final String NORMAL_QUEUE="normal_queue";
public static final String DEAD_QUEUE="dead_queue";
public static void main(String[] args) throws Exception {
Channel channel = RabbitUtils.rabbitConnection();
// 创建c1交换机
channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.TOPIC);
// 声明普通队列
HashMap map = new HashMap();
// 设置过期时间 10s 单位ms 这里有消费整去做控制
// map.put("x-message-ttl",100000);
// 正常队列设置死信交换机
map.put("x-dead-letter-exchange",DEAD_EXCHANGE);
// 设置死信的routerKey
map.put("x-dead-letter-routing-key","lisi");
// 创建普通队列
channel.queueDeclare(NORMAL_QUEUE,false,false,false,map);
//创建死信队列
channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
// 绑定
channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan");
channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi");
channel.basicConsume(NORMAL_QUEUE,true,(consumerTag, message) -> {
System.out.println("C1消息为："+message.getBody());
},cancelCallback->{
});
}
}
//消费者2
public class C2 {
public static final String DEAD_QUEUE="dead_queue";
private static final String DEAD_EXCHANGE="dead_exchange";
public static void main(String[] args) throws Exception {
Channel channel = RabbitUtils.rabbitConnection();
channel.basicConsume(DEAD_QUEUE,true,(consumerTag,message)->{
System.out.println("消息为："+new String(message.getBody()));
},cancelCallback->{});
}
}正常队列长度的限制：
根据c1做修改,测试报错先删除原来的队列与交换机
//设置正常队列长度的限制
map.put("x-max-length",6);
拒接消息:
添加手动应答拒接。
public class C1 {
private static final String NORMAL_EXCHANGE="normal_exchange";
private static final String DEAD_EXCHANGE="dead_exchange";
public static final String NORMAL_QUEUE="normal_queue";
public static final String DEAD_QUEUE="dead_queue";
public static void main(String[] args) throws Exception {
Channel channel = RabbitUtils.rabbitConnection();
// 创建c1交换机
channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.TOPIC);
// 声明普通队列
HashMap map = new HashMap();
// 设置过期时间 10s 单位ms
// map.put("x-message-ttl",100000);
// 正常队列设置死信交换机
map.put("x-dead-letter-exchange",DEAD_EXCHANGE);
// 设置死信的routerKey
map.put("x-dead-letter-routing-key","lisi");
// 设置正常队列长度的限制
// map.put("x-max-length",6);
// 创建普通队列
channel.queueDeclare(NORMAL_QUEUE,false,false,false,map);
//创建死信队列
channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
// 绑定
channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan");
channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi");
channel.basicConsume(NORMAL_QUEUE,false,(consumerTag, message) -> {
String msg = new String(message.getBody());
System.out.println("C1消息为："+msg);
// 拒接对应消息
if(msg.equals("info2")){
// deliveryTag
channel.basicReject(message.getEnvelope().getDeliveryTag(),false);
}else{
channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
}
},cancelCallback->{
});
}
}SpringAMQP
官网地址：Spring AMQP
Spring AMQP 是 Spring 框架中的一个模块，它提供了基于 AMQP（Advanced Message Queuing Protocol，高级消息队列协议）标准的抽象层，用于简化在 Spring 应用程序中使用消息队列的过程。Spring AMQP 不仅简化了与消息代理（如 RabbitMQ）的集成，还提供了一套高度可配置的模板类来生产、消费消息，并管理AMQP基础设施组件，如队(Queue)、交换机(Exchange)和绑定(Binding)。
使用
org.springframework.boot
spring-boot-starter-amqp
生产者
logging:
pattern:
dateformat: MM-dd HH:mm:ss:SSS
spring:
rabbitmq:
host: 38.6.217.70
port: 5672
username: itcast
password: 123321
virtual-host: /package cn.itcast.mq.spring;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
import javax.annotation.Resource;
@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {
@Resource
private RabbitTemplate rabbitTemplate;
@Test
public void testSendMessage2SimpleQueue() throws InterruptedException {
// 1.发送消息
String message = "Hello, Spring Amqp!";
rabbitTemplate.convertAndSend("simple.queue", message);
}
}消费者
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
//监听机制
@Component
public class SpringRabbitListener {
@RabbitListener(queues = "simple.queue")
public void listenSimpleQueueMessage(String msg) {
System.out.println("spring接收到的消息是：" + msg);
}
}预取限制
案例：将50条消息在一秒内分类交给两个消费者消费。
//生成者
@Test
public void testSendWordSimpleQueue() throws InterruptedException {
// 1.发送消息
String key ="simple.queue";
String message = "Hello, Spring Amqp____";
for (int i = 0; i < 49; i++) {
rabbitTemplate.convertAndSend(key, message+i);
Thread.sleep(20);
}
}
//消费者
@Component
public class SpringRabbitListener {
@RabbitListener(queues = "simple.queue")
public void listenSimpleQueueMessage(String msg) throws InterruptedException {
System.out.println("spring接收到的消息是：" + msg+"___"+ LocalDateTime.now());
Thread.sleep(20);
}
@RabbitListener(queues = "simple.queue")
public void listenFanoutQueue1(String msg) throws InterruptedException {
System.err.println("FanoutQueue1接收到的消息是：" + msg+"___"+ LocalDateTime.now());
Thread.sleep(200); //模拟性能
}
}通过执行结果我们可以看出listenFanoutQueue1这个监听器执行的是奇数，而listenSimpleQueueMessage则是偶数。且时间超出了1秒。为什么呢？
因为在生产者发送到队列中时，消费者会预取消息，在默认情况下进行平分机制，在上面代码中我们可以看到我们使用了线程睡眠的方式模拟了性能，在平分的情况下，睡眠200的执行了25条，所以导致了超出了1s。 如何调整呢？
logging:
pattern:
dateformat: MM-dd HH:mm:ss:SSS
spring:
rabbitmq:
host: 38.6.217.70
port: 5672
username: itcast
password: 123321
virtual-host: /
listener: #设置预取
simple:
prefetch: 1 #每次只取一条
#这段配置的作用是在使用 RabbitMQ 的时候，配置消费者监听器的简单模式，并设置消息预取值为 1。这意味着每次只会从队列中取出一条消息进行处理，处理完后再去取下一条消息。这种方式可以保证消息的顺序处理。文章来源:https://blog.csdn.net/m0_64787068/article/details/140063946
微信扫描下方的二维码阅读本文
© 版权声明
THE END
暂无评论内容