个人中心
文章发布
退出
作者热门文章
- Java锁的逻辑（结合对象头和ObjectMonitor）
- 还在用饼状图？来瞧瞧这些炫酷的百分比可视化新图形（附代码实现）⛵
- 自动注册实体类到EntityFrameworkCore上下文，并适配ABP及ABPVNext
- 基于Sklearn机器学习代码实战
37
4
1、MQ全称为Message Queue,消息队列（MQ）是⼀种应⽤程序对应⽤程序的通信⽅法。应⽤程序通过读写出⼊队列的消息（针对应⽤程序的数据）来通信，⽽⽆需专⽤连接来链接它们。2、消息传递指的是程序之间通过在消息中发送数据进⾏通信，⽽不是通过直接调⽤彼此来通信，直接调⽤通常是⽤于诸如远程过程调⽤的技术。
1、RabbitMQ 稳定可靠,数据⼀致,⽀持多协议,有消息确认,基于erlang语⾔2、Kafka ⾼吞吐,⾼性能,快速持久化,⽆消息确认,⽆消息遗漏,可能会有有重复消息,依赖于zookeeper,成本⾼.3、ActiveMQ 不够灵活轻巧,对队列较多情况⽀持不好.4、RocketMQ 性能好,⾼吞吐,⾼可⽤性,⽀持⼤规模分布式，协议⽀持单⼀
1、RabbitMQ是⼀个在AMQP基础上完成的，可复⽤的企业消息系统。他遵循MozillaPublic License开源协议。2、AMQP，即Advanced Message Queuing Protocol, ⼀个提供统⼀消息服务的应⽤层标准⾼级消息队列协议,是应⽤层协议的⼀个开放标准,为⾯向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同的开发语⾔等条件的限制。Erlang中的实现有 RabbitMQ等。3、主要特性：
- 保证可靠性 ：使⽤⼀些机制来保证可靠性，如持久化、传输确认、发布确认
- 灵活的路由功能
- 可伸缩性：⽀持消息集群，多台RabbitMQ服务器可以组成⼀个集群
- ⾼可⽤性 ：RabbitMQ集群中的某个节点出现问题时队列仍然可⽤
- ⽀持多种协议
- ⽀持多语⾔客户端
- 提供良好的管理界⾯
- 提供跟踪机制：如果消息出现异常，可以通过跟踪机制分析异常原因
- 提供插件机制：可通过插件进⾏多⽅⾯扩展
具体参考： https://note.youdao.com/s/MKn2Jr8c 。
RabbitMQ默认提供了⼀个guests账号，但是此账号不能⽤作远程登录，也就是不能在管理系统的登录；我们可以创建⼀个新的账号并授予响应的管理权限来实现远程登录
⽤户虚拟主机队列
1、在linux中使⽤命令⾏创建⽤户 。
## 进⼊到rabbit_mq的sbin⽬录 cd /usr/local/rabbitmq_server- 3.7 . 0 / sbin ## 新增⽤户 . /rabbitmqctl add_user ytao admin1232、设置⽤户级别 。
## ⽤户级别： ## 1 .administrator 可以登录控制台、查看所有信息、可以对RabbitMQ进⾏管理 ## 2 .monitoring 监控者 登录控制台、查看所有信息 ## 3 .policymaker 策略制定者 登录控制台、指定策略 ## 4 .managment 普通管理员 登录控制台 . /rabbitmqctl set_user_tags ytao administrator
管理系统登录：访问http://localhost:15672/
RabbitMQ提供了多种消息的通信⽅式—⼯作模式 https://www.rabbitmq.com/getstarted.html消息通信是由两个⻆⾊完成：消息⽣产者（producer）和 消息消费者（Consumer）
⼀个队列只有⼀个消费者
多个消费者监听同⼀个队列
⼀个交换机绑定多个消息队列，每个消息队列有⼀个消费者监听
⼀个交换机绑定多个消息队列，每个消息队列都由⾃⼰唯⼀的key，每个消息队列有⼀个消费者监听
1、创建Maven项⽬ 。
2、添加RabbitMQ连接所需要的依赖 。
<!-- https: // mvnrepository.com/artifact/com.rabbitmq/amqp-client -- > <dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version> 4.10 . 0 </version> </dependency> <!-- https: // mvnrepository.com/artifact/org.slf4j/slf4j-log4j12 --> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-log4j12</artifactId> <version> 1.7 . 25 </version> <scope>test</scope> </dependency> <!-- https: // mvnrepository.com/artifact/org.apache.commons/commonslang3 --> <dependency> <groupId>org.apache.commons</groupId> <artifactId>commons-lang3</artifactId> <version> 3.9 </version> </dependency>3、在resources⽬录下创建log4j.properties 。
log4j.rootLogger=DEBUG,A1 log4j.logger.com.taotao = DEBUG log4j.logger.org.mybatis = DEBUG log4j.appender.A1 = org.apache.log4j.ConsoleAppender log4j.appender.A1.layout = org.apache.log4j.PatternLayout log4j.appender.A1.layout.ConversionPattern =%-d{yyyy-MM- dd HH:mm:ss,SSS} [ %t] [%c]-[%p] %m%n4、创建MQ连接工具类 。
import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import java.io.IOException; import java.util.concurrent.TimeoutException; public class ConnectionUtil { public static Connection getConnection() throws IOException, TimeoutException { // 1.创建连接⼯⼚ ConnectionFactory factory = new ConnectionFactory(); // 2.在⼯⼚对象中设置MQ的连接信息 (ip,port,virtualhost,username,password) factory.setHost( "47.96.11.185" ); factory.setPort( 5672 ); factory.setVirtualHost( "host1" ); factory.setUsername( "ytao" ); factory.setPassword( "admin123" ); // 3.通过⼯⼚对象获取与MQ的链接 Connection connection = factory.newConnection(); return connection; } }5、消息⽣产者发送消息 。
import com.qfedu.mq.utils.ConnectionUtil; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; public class SendMsg { public static void main(String[] args) throws Exception{ String msg = "Hello HuangDaoJun!" ; Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); // 定义队列(使⽤Java代码在MQ中新建⼀个队列) // 参数1：定义的队列名称 // 参数2：队列中的数据是否持久化（如果选择了持久化） // 参数3: 是否排外（当前队列是否为当前连接私有） // 参数4：⾃动删除（当此队列的连接数为0时，此队列会销毁（⽆论队列中是否 还有数据）） // 参数5：设置当前队列的参数 // channel.queueDeclare("queue7",false,false,false,null); // 参数1：交换机名称，如果直接发送信息到队列，则交换机名称为"" // 参数2：⽬标队列名称 // 参数3：设置当前这条消息的属性（设置过期时间 10） // 参数4：消息的内容 channel.basicPublish("","queue1", null ,msg.getBytes()); System.out.println( "发送：" + msg); channel.close(); connection.close(); } }
1、创建Maven项⽬2、添加依赖3、log4j.properties4、ConnetionUtil.java5、消费者消费消息import com.qfedu.mq.utils.ConnectionUtil; import com.rabbitmq.client.* ; import java.io.IOException; import java.util.concurrent.TimeoutException; public class ReceiveMsg { public static void main(String[] args) throws IOException, TimeoutException { Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); Consumer consumer = new DefaultConsumer(channel){ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte [] body) throws IOException { // body就是从队列中获取的数据 String msg = new String(body); System.out.println( "接收："+ msg); } }; channel.basicConsume( "queue1", true ,consumer); } }
⼀个发送者多个消费者
public
class
SendMsg {
public
static
void
main(String[] args)
throws
Exception{
System.out.println(
"请输⼊消息："
);
Scanner scanner
=
new
Scanner(System.in);
String msg
=
null
;
while
(!"quit".equals(msg =
scanner.nextLine())){
Connection connection
=
ConnectionUtil.getConnection();
Channel channel
=
connection.createChannel();
channel.basicPublish(
"","queue2",
null
,msg.getBytes());
System.out.println(
"发送：" +
msg);
channel.close();
connection.close();
}
}
}
public
class
ReceiveMsg {
public
static
void
main(String[] args)
throws
Exception {
Connection connection
=
ConnectionUtil.getConnection();
Channel channel
=
connection.createChannel();
Consumer consumer
=
new
DefaultConsumer(channel){
@Override
public
void
handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
byte
[] body)
throws
IOException {
//
body就是从队列中获取的数据
String msg =
new
String(body);
System.out.println(
"Consumer1接收："+
msg);
if
("wait"
.equals(msg)){
try
{
Thread.sleep(
10000
);
}
catch
(InterruptedException e) {
e.printStackTrace();
}
}
}
};
channel.basicConsume(
"queue2",
true
,consumer);
}
}
public
class
ReceiveMsg {
public
static
void
main(String[] args)
throws
IOException,
TimeoutException {
Connection connection
=
ConnectionUtil.getConnection();
Channel channel
=
connection.createChannel();
Consumer consumer
=
new
DefaultConsumer(channel){
@Override
public
void
handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
byte
[] body)
throws
IOException {
//
body就是从队列中获取的数据
String msg =
new
String(body);
System.out.println(
"Consumer2接收："+
msg);
}
};
channel.basicConsume(
"queue2",
true
,consumer);
}
}
public
class
SendMsg {
public
static
void
main(String[] args)
throws
Exception{
System.out.println(
"请输⼊消息："
);
Scanner scanner
=
new
Scanner(System.in);
String msg
=
null
;
while
(!"quit".equals(msg =
scanner.nextLine())){
Connection connection
=
ConnectionUtil.getConnection();
Channel channel
=
connection.createChannel();
channel.basicPublish(
"ex1","",
null
,msg.getBytes());
System.out.println(
"发送：" +
msg);
channel.close();
connection.close();
}
}
}
public
class
ReceiveMsg1 {
public
static
void
main(String[] args)
throws
Exception {
Connection connection
=
ConnectionUtil.getConnection();
Channel channel
=
connection.createChannel();
Consumer consumer
=
new
DefaultConsumer(channel){
@Override
public
void
handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
byte
[] body)
throws
IOException {
//
body就是从队列中获取的数据
String msg =
new
String(body);
System.out.println(
"Consumer1接收："+
msg);
if
("wait"
.equals(msg)){
try
{
Thread.sleep(
10000
);
}
catch
(InterruptedException e) {
e.printStackTrace();
}
}
}
};
channel.basicConsume(
"queue3",
true
,consumer);
}
}
public
class
ReceiveMsg2 {
public
static
void
main(String[] args)
throws
IOException,
TimeoutException {
Connection connection
=
ConnectionUtil.getConnection();
Channel channel
=
connection.createChannel();
Consumer consumer
=
new
DefaultConsumer(channel){
@Override
public
void
handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
byte
[] body)
throws
IOException {
//
body就是从队列中获取的数据
String msg =
new
String(body);
System.out.println(
"Consumer2接收："+
msg);
}
}
;
channel.basicConsume(
"queue4",
true
,consumer);
}
}
public
class
SendMsg {
public
static
void
main(String[] args)
throws
Exception{
System.out.println(
"请输⼊消息："
);
Scanner scanner
=
new
Scanner(System.in);
String msg
=
null
;
while
(!"quit".equals(msg =
scanner.nextLine())){
Connection connection
=
ConnectionUtil.getConnection();
Channel channel
=
connection.createChannel();
if
(msg.startsWith("a"
)){
channel.basicPublish(
"ex2","a",
null
,msg.getBytes());
}
else
if
(msg.startsWith("b"
)){
channel.basicPublish(
"ex2","b",
null
,msg.getBytes());
}
System.out.println(
"发送：" +
msg);
channel.close();
connection.close();
}
}
}
public
class
ReceiveMsg1 {
public
static
void
main(String[] args)
throws
Exception {
Connection connection
=
ConnectionUtil.getConnection();
Channel channel
=
connection.createChannel();
Consumer consumer
=
new
DefaultConsumer(channel){
@Override
public
void
handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
byte
[] body)
throws
IOException {
//
body就是从队列中获取的数据
String msg =
new
String(body);
System.out.println(
"Consumer1接收："+
msg);
if
("wait"
.equals(msg)){
try
{
Thread.sleep(
10000
);
}
catch
(InterruptedException e) {
e.printStackTrace();
}
}
}
};
channel.basicConsume(
"queue5",
true
,consumer);
}
}
public
class
ReceiveMsg2 {
public
static
void
main(String[] args)
throws
IOException, TimeoutException {
Connection connection
=
ConnectionUtil.getConnection();
Channel channel
=
connection.createChannel();
Consumer consumer
=
new
DefaultConsumer(channel){
@Override
public
void
handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
byte
[] body)
throws
IOException {
//
body就是从队列中获取的数据
String msg =
new
String(body);
System.out.println(
"Consumer2接收："+
msg);
}
};
channel.basicConsume(
"queue6",
true
,consumer);
}
}
SpringBoot应⽤可以完成⾃动配置及依赖注⼊——可以通过Spring直接提供与MQ的连接对象
1、创建SpringBoot应⽤，添加依赖 。
2、配置application.yml 。
server: port: 9001 spring: application: name: producer rabbitmq: host: 47.96.11.185 port: 5672 virtual - host: host1 username: ytao password: admin1233、发送消息 。
@Service public class TestService { @Resource private AmqpTemplate amqpTemplate; public void sendMsg(String msg){ // 1. 发送消息到队列 amqpTemplate.convertAndSend("queue1" ,msg); // 2. 发送消息到交换机(订阅交换机) amqpTemplate.convertAndSend("ex1","" ,msg); // 3. 发送消息到交换机(路由交换机) amqpTemplate.convertAndSend("ex2","a" ,msg); } }
1、创建项⽬添加依赖2、配置yml3、接收消息@Service // @RabbitListener(queues = {"queue1","queue2"}) @RabbitListener(queues = "queue1" ) public class ReceiveMsgService { @RabbitHandler public void receiveMsg(String msg){ System.out.println( "接收MSG："+ msg); } }
RabbitMQ是消息队列，发送和接收的都是字符串/字节数组类型的消息
要求：传递的对象实现序列化接⼝传递的对象的包名、类名、属性名必须⼀致
。
1、消息提供者 。
@Service public class MQService { @Resource private AmqpTemplate amqpTemplate ; public void sendGoodsToMq(Goods goods){ // 消息队列可以发送 字符串、字节数组、序列化对象 amqpTemplate.convertAndSend("","queue1" ,goods); } }2、消息消费者 。
@Component @RabbitListener(queues = "queue1") public class ReceiveService { @RabbitHandler public void receiveMsg(Goods goods){ System.out.println( "Goods---"+ goods); } }
要求：传递的对象实现序列化接⼝传递的对象的包名、类名、属性名必须⼀致
。
1、消息提供者 。
@Service public class MQService { @Resource private AmqpTemplate amqpTemplate; public void sendGoodsToMq(Goods goods){ // 消息队列可以发送 字符串、字节数组、序列化对象 byte[] bytes = SerializationUtils.serialize(goods); amqpTemplate.convertAndSend( "","queue1" ,bytes); } }2、消息消费者 。
@Component @RabbitListener(queues = "queue1" ) public class ReceiveService { @RabbitHandler public void receiveMsg( byte [] bs){ Goods goods = (Goods) SerializationUtils.deserialize(bs); System.out.println( "byte[]---"+ goods); } }
要求：对象的属性名⼀直
。
1、消息提供者 。
@Service public class MQService { @Resource private AmqpTemplate amqpTemplate; public void sendGoodsToMq(Goods goods) throws JsonProcessingException { // 消息队列可以发送 字符串、字节数组、序列化对象 ObjectMapper objectMapper = new ObjectMapper(); String msg = objectMapper.writeValueAsString(goods); amqpTemplate.convertAndSend( "","queue1" ,msg); } }2、消息消费者 。
@Component @RabbitListener(queues = "queue1" ) public class ReceiveService { @RabbitHandler public void receiveMsg(String msg) throws JsonProcessingException { ObjectMapper objectMapper = new ObjectMapper(); Goods goods = objectMapper.readValue(msg,Goods.class ); System.out.println( "String---"+ msg); } }
我们使⽤消息队列，消息队列和交换机可以通过管理系统完成创建，也可以在应⽤程序中通过Java代码来完成创建
1、使⽤Java代码新建队列 。
// 1.定义队列 (使⽤Java代码在MQ中新建⼀个队列) // 参数1：定义的队列名称 // 参数2：队列中的数据是否持久化（如果选择了持久化） // 参数3: 是否排外（当前队列是否为当前连接私有） // 参数4：⾃动删除（当此队列的连接数为0时，此队列会销毁（⽆论队列中是否还有数 据）） // 参数5：设置当前队列的参数 channel.queueDeclare("queue7", false , false , false , null );2、新建交换机 。
// 定义⼀个“订阅交换机” channel.exchangeDeclare("ex3" , BuiltinExchangeType.FANOUT); // 定义⼀个“路由交换机” channel.exchangeDeclare("ex4", BuiltinExchangeType.DIRECT);3、绑定队列到交换机 。
// 绑定队列 // 参数1：队列名称 // 参数2：⽬标交换机 // 参数3：如果绑定订阅交换机参数为"",如果绑定路由交换机则表示设置队列的key channel.queueBind("queue7","ex4","k1" ); channel.queueBind( "queue8","ex4","k2");
@Configuration
public
class
RabbitMQConfiguration {
//
声明队列
@Bean
public
Queue queue9(){
Queue queue9
=
new
Queue("queue9"
);
//
设置队列属性
return
queue9;
}
@Bean
public
Queue queue10(){
Queue queue10
=
new
Queue("queue10"
);
//
设置队列属性
return
queue10;
}
//
声明订阅模式交换机
@Bean
public
FanoutExchange ex5(){
return
new
FanoutExchange("ex5"
);
}
//
声明路由模式交换机
@Bean
public
DirectExchange ex6(){
return
new
DirectExchange("ex6"
);
}
//
绑定队列
@Bean
public
Binding bindingQueue9(Queue queue9, DirectExchange ex6){
return
BindingBuilder.bind(queue9).to(ex6).with("k1"
);
}
@Bean
public
Binding bindingQueue10(Queue queue10, DirectExchange ex6){
return
BindingBuilder.bind(queue10).to(ex6).with("k2"
);
}
}
消息的可靠性：从 ⽣产者发送消息 —— 消息队列存储消息 —— 消费者消费消息 的整个过程中消息的安全性及可控性。
- ⽣产者
- 消息队列
- 消费者
RabbitMQ事务指的是基于客户端实现的事务管理，当在消息发送过程中添加了事务，处理效率降低⼏⼗倍甚⾄上百倍
Connection connection = RabbitMQUtil.getConnection();
//
connection 表示与 host1的连接
Channel channel =
connection.createChannel();
channel.txSelect();
//
开启事务
try
{
channel.basicPublish(
"ex4", "k1",
null
, msg.getBytes());
channel.txCommit();
//
提交事务
}
catch
(Exception e){
channel.txRollback();
//
事务回滚
}
finally
{
channel.close();
connection.close();
}
1、消息确认机制：确认消息提供者是否成功发送消息到交换机2、return机制：确认消息是否成功的从交换机分发到队列
1、普通confirm⽅式 。
// 1.发送消息之前开启消息确认 channel.confirmSelect(); channel.basicPublish( "ex1", "a", null , msg.getBytes()); // 2.接收消息确认 Boolean b = channel.waitForConfirms(); System.out.println( "发送：" +(b?"成功":"失败"));2、批量confirm⽅式 。
// 1.发送消息之前开启消息确认 channel.confirmSelect(); // 2.批量发送消息 for ( int i=0 ; i<10 ; i++ ){ channel.basicPublish( "ex1", "a", null , msg.getBytes()); } // 3.接收批量消息确认：发送的所有消息中，如果有⼀条是失败的，则所有消息发送直接失败，抛出IO异常 Boolean b = channel.waitForConfirms();3、异步confirm⽅式 。
// 发送消息之前开启消息确认 channel.confirmSelect(); // 批量发送消息 for ( int i=0 ; i<10 ; i++ ){ channel.basicPublish( "ex1", "a", null , msg.getBytes()); } // 假如发送消息需要10s，waitForConfirms会进⼊阻塞状态 // boolean b = channel.waitForConfirms(); // 使⽤监听器异步confirm channel.addConfirmListener ( new ConfirmListener() { // 参数1： long l 返回消息的表示 // 参数2： boolean b 是否为批量confirm public void handleAck( long l, Boolean b) throws IOException { System.out.println( "~~~~~消息成功发送到交换机" ); } public void handleNack( long l, Boolean b) throws IOException { System.out.println( "~~~~~消息发送到交换机失败" ); } } );
1、添加return监听器2、发送消息是指定第三个参数为true3、由于监听器监听是异步处理，所以在消息发送之后不能关闭channelString msg = "Hello HuangDaoJun!" ; Connection connection = ConnectionUtil.getConnection(); // 相当于JDBC操作的数据库连接 Channel channel = connection.createChannel(); // 相当于JDBC操作的statement // return机制：监控交换机是否将消息分发到队列 channel.addReturnListener( new ReturnListener() { public void handleReturn( int i, String s, String s1, String s2,AMQP.BasicProperties basicProperties, byte [] bytes) throws IOException { // 如果交换机分发消息到队列失败，则会执⾏此⽅法（⽤来处理交换机分发消息到队列失败的情况） System.out.println("*****"+ i); // 标识 System.out.println("*****"+ s); // System.out.println("*****"+ s1); // 交换机名 System.out.println("*****"+ s2); // 交换机对应的队列的key System.out.println("*****"+ new String(bytes)); // 发送的消息 } } ); // 发送消息 // channel.basicPublish("ex2", "c", null, msg.getBytes()); channel.basicPublish("ex2", "c", true , null , msg.getBytes());
1、配置application.yml,开启消息确认和return监听 。
spring: rabbitmq: publisher -confirm- type: simple ## 开启消息确认模式 publisher -returns: true ##使⽤return监听机制2、创建confirm和return监听 。
2.1、消息确认 。
@Component public class MyConfirmListener implements RabbitTemplate.ConfirmCallback { @Autowired private AmqpTemplate amqpTemplate; @Autowired private RabbitTemplate rabbitTemplate; @PostConstruct public void init(){ rabbitTemplate.setConfirmCallback( this ); } @Override public void confirm (CorrelationData correlationData, Boolean b, String s) { // 参数b 表示消息确认结果 // 参数s 表示发送的消息 if (b){ System.out.println( "消息发送到交换机成功！" ); } else { System.out.println( "消息发送到交换机失败！" ); amqpTemplate.convertAndSend( "ex4","" ,s); } } }2.2、return机制 。
@Component public class MyReturnListener implements RabbitTemplate.ReturnsCallback { @Autowired private AmqpTemplate amqpTemplate; @Autowired private RabbitTemplate rabbitTemplate; @PostConstruct public void init(){ rabbitTemplate.setReturnsCallback( this ); } @Override public void returnedMessage (ReturnedMessage returnedMessage) { System.out.println( "消息从交换机分发到队列失败" ); String exchange = returnedMessage.getExchange(); String routingKey = returnedMessage.getRoutingKey(); String msg = returnedMessage.getMessage().toString(); amqpTemplate.convertAndSend(exchange,routingKey,msg); } }
@Component
@ Ra bbitListener
(queues
="queue01"
)
public
class
Consumer1 {
@RabbitHandler
public
void
process(String msg,Channel channel, Message message)
throws
IOException {
try
{
System.out.println(
"get msg1 success msg = "+
msg);
/**
* 确认⼀条消息：<br>
* channel.basicAck(deliveryTag, false); <br>
* deliveryTag:该消息的index <br>
* multiple：是否批量.true:将⼀次性ack所有⼩于deliveryTag的消息 <br>
*/
channel.basicAck(message.getMessageProperties().getDeliveryTag(),
false
);
}
catch
(Exception e) {
//
消费者处理出了问题，需要告诉队列信息消费失败
/**
* 拒绝确认消息:<br>
* channel.basicNack(long deliveryTag, boolean multiple, boolean requeue) ; <br>
* deliveryTag:该消息的index<br>
* multiple：是否批量.true:将⼀次性拒绝所有⼩于deliveryTag的消息。<br>
* requeue：被拒绝的是否重新⼊队列 <br>
*/
channel.basicNack(message.getMessageProperties().getDeliveryTag(),
false
,
true
);
System.err.println(
"get msg1 failed msg = "+
msg);
}
}
}
消息消费的幂等性——多次消费的执⾏结果时相同的 （避免重复消费）解决⽅案：处理成功的消息setnx到redis
1、延迟队列——消息进⼊到队列之后，延迟指定的时间才能被消费者消费2、AMQP协议和RabbitMQ队列本身是不⽀持延迟队列功能的，但是可以通过TTL（Time To Live）特性模拟延迟队列的功能3、TTL就是消息的存活时间。RabbitMQ可以分别对队列和消息设置存活时间
1、在创建队列的时候可以设置队列的存活时间，当消息进⼊到队列并且在存活时间内没有消费者消费，则此消息就会从当前队列被移除；2、创建消息队列没有设置TTL，但是消息设置了TTL，那么当消息的存活时间结束，也会被移除；3、当TTL结束之后，我们可以指定将当前队列的消息转存到其他指定的队列
1、实现流程图 。
2、创建交换机和队列 。
1、解耦 。
场景说明：⽤户下单之后，订单系统要通知库存系统
2、异步 。
场景说明：⽤户注册成功之后，需要发送注册邮件及注册短信提醒
3、消息通信 。
场景说明：应⽤系统之间的通信，例如聊天室
4、流量削峰 。
场景说明：秒杀业务
5、⽇志处理 。
场景说明：系统中⼤量的⽇志处理
。
最后此篇关于RabbitMQ保姆级教程最佳实践的文章就讲到这里了,如果你想了解更多关于RabbitMQ保姆级教程最佳实践的内容请搜索CFSDN的文章或继续浏览相关文章，希望大家以后支持我的博客！ 。
37
4
0
本文分享自华为云社区《大模型LLM之分布式训练》，作者： 码上开花_Lancer。 随着语言模型参数量和所需训练数据量的急速增长，单个机器上有限的资源已无法满足大语言模型训练的要求。需要设计分布式训
本文分享自华为云社区《五大基础算法--动态规划法》，作者： 大金（内蒙的）。 一、基本概念 动态规划法，和分治法极其相似。区别就是，在求解子问题时，会保存该子问题的解，后面的子问题求解时，可以直接拿来
pip install scp pip install pexpect 测试代码： import os import stat import paramiko # 用于调用scp命令 def s
我目前正在实现“ token ”REST 服务。 token 只是一个字符串，由一些参数构建而成，然后经过哈希处理并在一定时间后过期。 我想在我的 REST 服务中有一个可以验证 token 的端点，
打开软删除后，我在客户端上添加一条记录，推送，删除添加的记录推送，然后尝试使用与初始记录相同的主键添加新记录(然后推送)，我得到一个异常(exception)。 EntityDomainManager
打开软删除后，我在客户端上添加一条记录，推送，删除添加的记录推送，然后尝试使用与初始记录相同的主键添加新记录(然后推送)，我得到一个异常(exception)。 EntityDomainManager
我有一个应用程序，每 x 秒接收一次天气信息。我想将此数据保存到 XML 文件中。 我应该为每个天气通知创建一个新的 XML 文件，还是将每个通知附加到同一个 XML 文件中？我不确定 XML 标准的
我猜我们大多数人都必须在某个时候处理这个问题，所以我想我会问这个问题。 当您的 BLL 中有很多集合并且您发现自己一遍又一遍地编写相同的旧内联(匿名)谓词时，显然有必要进行封装，但实现封装的最佳方
我有一些 c# 代码已经运行了一段时间了..我不得不说，虽然我了解 OO 原则的基础知识，但显然有不止一种方法可以给猫剥皮(尽管我讨厌那个短语!)。 因此，我有一个基本抽象类作为基本数据服务类，如下所
我设计了一个 SQL 数据库系统(使用 Postgre)，我有一个问题，即创建一个关系/引用的常见做法是什么，这种关系/引用即使在引用的对象被删除时也能持续存在。 比如有一个UserORM，还有Act
我们的目标是搜索用户输入的字符串并计算在其中找到多少元音。不幸的是我被困在这里，有什么帮助吗？ def numVowels(s): vowels= "AEIOUaeiou" if s
我有一个适用于我的“items”int 数组的旋转函数。下面的代码完成了它，除了我不必要地传输值。我正在努力实现“就地”轮换。我的意思是 ptrs 会递增或递减，而不是从数组中获取值。我需要通过这种方
我有一个 json 存储在我的应用程序文档文件夹中，我需要在我的所有 View 中使用它。我正在加载 json 并将其添加到每个 View 中的 NSMutableArray。但现在我了解到，我可以将
我用 C++ 开始了一个项目。这种语言的内存管理对我来说是新的。 我过去常常使用 new () 创建对象，然后传递指针，虽然它可以工作，但调试起来很痛苦，人们看到代码时会用有趣的眼神看着我。我为它没有
已结束。 这个问题是 off-topic .它目前不接受答案。 想要改进这个问题？ Update the question所以它是on-topic堆栈溢出。 关闭 10 年前。 Improve thi
保持类松散耦合是编写易于理解、修改和调试的代码的一个重要方面——我明白这一点。然而，作为一个新手，几乎任何时候我都会超越我所苦苦挣扎的最简单的例子。 我或多或少地了解如何将字符串、整数和简单数据类型封
我发现我需要编写大量重复代码，因为我无法从其他 Controller 调用函数。例如，这里新闻提要内容在我的代码中重复，我对一个 Controller 做一些特定的事情，然后需要像这样加载我的新闻提要
假设需要一种数字数据类型，其允许值在指定范围内。更具体地说，假设要定义一个整数类型，其最小值为0，最大值为5000。这种情况在很多情况下都会出现，例如在对数据库数据类型，XSD数据类型进行建模时。 在
假设我想循环整个数组来访问每个元素。使用 for 循环、for...in 循环或 for...of 循环是 JavaScript 开发人员的标准做法吗？ 例如: var myArray = ["app
我有一个旧的 SL4/ria 应用程序，我希望用 Breeze 取代它。我有一个关于内存使用和缓存的问题。我的应用程序加载工作列表(一个典型的用户可以访问大约 1,000 个这些工作)。此外，还有很多
我是一名优秀的程序员,十分优秀！