RabbitMQ学习笔记（四、RabbitMQ队列）
目录：
- 消息路由失败了会怎样
- 备份交换器
- TTL与DLX
- 如何实现延迟队列
- RabbitMQ的RPC实现
- 持久化
- 事务
- 发送方确认机制
消息路由失败了会怎样：
在RabbitMQ中，如果消息路由失败了，一般会有两种情况。要么是把消息回退给客户端处理，要么就把消息丢弃。
处理逻辑是根据basicPublish方法的mandatory和immediate两个参数来控制。
1、mandatory：当mandatory=true时，如果交换器无法根据自身类型和路由键匹配到符合条件的队列，便会调用Basic.Return命令将消息会推给生产者；当mandatory=false时，不满足条件则丢弃此条消息。
channel.addReturnListener(new ReturnListener() {
public void handleReturn(int replyCode, String replyText, String exchange, String routingKey,
AMQP.BasicProperties properties, byte[] body) throws IOException {
// 具体处理逻辑
}
});
2、immediate：当immediate=true时，交换器将消息路由到队列后，发现此队列上不存在任何消费者，那么这条消息将不会放入到队列中。当路由键匹配的所有队列都没有消费者时，改消息将会通过Basic.Return返回给生产者。
备份交换器：
备份交换器可以将未被路由到的消息存储在RabbitMQ中，在需要它的时候再去使用。
public class AlternateProduct {
private static final String EXCHANGE_NAME = "alternate.exchange";
private static final String EXCHANGE_BAK_NAME = "alternate-bak.exchange";
private static final String QUEUE_NAME = "alternate.queue";
private static final String QUEUE_BAK_NAME = "alternate-bak.queue";
private static final String ROUTING_KEY_NAME = "alternate.routing.key";
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = RabbitMqUtils.getConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT, false, false, false, getExchangeDeclareArgs());
// fanout类型，放款路由限制
channel.exchangeDeclare(EXCHANGE_BAK_NAME, BuiltinExchangeType.FANOUT, false, false, false, null);
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
channel.queueDeclare(QUEUE_BAK_NAME, false, false, false, null);
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY_NAME);
// 因为交换器QUEUE_BAK_NAME设置fanout类型，所以可以不必关心路由键，故随便写可能将消息路由到对应的队列中
channel.queueBind(QUEUE_BAK_NAME, EXCHANGE_BAK_NAME, "123");
// 发消息时路由键设置一个不存在的""，让其路由不到，从而把消息发到备份队列中
channel.basicPublish(EXCHANGE_NAME, "", MessageProperties.PERSISTENT_TEXT_PLAIN,
"alternate".getBytes());
RabbitMqUtils.close(connection, channel);
}
private static Map<String, Object> getExchangeDeclareArgs() {
Map<String, Object> result = new HashMap<String, Object>(1);
result.put("alternate-exchange", EXCHANGE_BAK_NAME);
return result;
}
}
关于备份交换器的注意点：
1、如果备份交换器不存在，客户端和RabbitMQ客户端都不会出现异常，但是消息会丢失。
2、如果备份交换器没有绑定任何队列，客户端和RabbitMQ客户端都不会出现异常，但是消息会丢失。
3、如果备份交换器没有匹配到任何队列，客户端和RabbitMQ客户端都不会出现异常，但是消息会丢失。
4、如果备份交换器和mandatory一起使用，且备份交换器有效，此时mandatory将无效。
TTL与DLX：
1、TTL：过期时间，有队列过期时间和消息过期时间。
）队列过期时间：
通过设置队列的过期时间，来使队列中所以的消息都具有过期时间。
）消息过期时间：
设置消息的BasicProperties props属性值来控制消息的过期时间。
AMQP.BasicProperties.Builder publishBuilder = new AMQP.BasicProperties.Builder();
// expiration单位ms
publishBuilder.expiration("10000"); channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY_NAME, publishBuilder.build(),
"ttl".getBytes());
对于第一种TTL来说，队列一但过期就会删除调；但对于第二种TTL来说，队列过期不会马上删除，而是等队列要被消费时再判断是否要删除。
那为什么会不一样呢，我们都知道mq对性能的要求是非常高的，如果第二种ddl的方式也要及时删除的话势必要扫描整个队列，这样的话，若队列长度较大是性能便会非常的差。
而第一种为什么可以做到及时删除呢，我们知道队列具有先进先出的特性，所以先入队的肯定要比后入队的要先过期，所以只要删除头部的就好啦。
而第二种的消息过期时间都是不固定的，考虑到MQ的性能，所以采用了上述的方式。
2、DLX：死信交换器，全称Dead Letter Exchange
变为死信队列的有以下几种情况：
- 消息被拒，且requeue=false
- 队列过期或队列达到最大长度
注意：DLX也是一个正常的交换器，和一般队列没有区别，它能在任何的队列上被指定。
如何实现延迟队列：
本模块讲述RabbitMQ，仅提供RabbitMQ的实现，大佬们有兴趣可以实现其它几种方式。
延迟队列是指将消息发送到队列后，等待一段时间后再进行消费。场景：饿了么外卖下单后，超过15分钟订单失效。
延迟队列场景的时间方式有四种：
1、DB轮询：通过job或其它逻辑将订单表的必要字段查出（如：orderId、createTime、status），当订单超过xx时间，将状态置为失效。
）优点：实现简单、无技术难点、异常恢复、支持分布式/集群环境
）缺点：影响DB性能、时效性查、效率低
2、JDK DelayQueue：java api提供的延迟队列的实现，通过poll()、take()方法获取超时任务
）优点：实现简单、性能较好
）缺点：异常恢复困难、分布式/集群实现困难（基于JVM内存）
3、Redis sortedSet：通过zset类型的score来实现
）优点：解耦、异常恢复、扩展性强、支持分布式/集群环境
）缺点：增加了redis维护成本、占用带宽
4、RabbitMQ TTL + DLX：使用RabbitMQ的过期时间和死信队列实现
）优点：解耦、异常恢复、扩展性强、支持分布式/集群环境
）缺点：增加了RabbitMQ维护成本、占用带宽
RabbitMQ的RPC实现：
RabbitMQ也可以实现RPC，客户端发送消息，服务端接收消息。
replayTo：设置回调队列，用于客户端响应服务端的回调消息。
correlationId：RPC请求和响应的关联id。
public class RpcServer {
private static final String QUEUE_NAME = "rpc.queue";
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = RabbitMqUtils.getRpcConnection();
final Channel channel = connection.createChannel();
// 创建请求处理队列，用于服务端接收客户端RPC请求
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
System.out.println("等待RPC请求...");
// 服务端监听客户端发送的RPC请求
channel.basicConsume(QUEUE_NAME, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
throws IOException {
String correlationId = properties.getCorrelationId();
String message = "";
try {
message = new String(body);
System.err.println(format("service recv message:{0}, corrId:{1}", message, correlationId));
} catch (Exception e) {
e.printStackTrace();
} finally {
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
.correlationId(correlationId)
.build();
// 使用默认exchange，允许通过routingKey指定message将被发送给哪个queue
channel.basicPublish("", properties.getReplyTo(), props, (message + "--is done.").getBytes("UTF-8"));
channel.basicAck(envelope.getDeliveryTag(), false);
}
}
});
}
}
public class RpcClient {
private static final String QUEUE_NAME = "rpc.queue";
public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
final Connection connection = RabbitMqUtils.getConnection();
Channel channel = connection.createChannel();
// 随机创建corrId
final String collId = UUID.randomUUID().toString();
// 客户端创建匿名队列，用于响应服务端请求
String callbackQueueName = channel.queueDeclare().getQueue();
// 客户端发送消息；使用默认exchange（exchange=""），允许通过routingKey指定message将被发送给哪个queue
channel.basicPublish("", QUEUE_NAME, getBasicPublishProperties(collId, callbackQueueName),
"hello world".getBytes());
// 客户端接收服务端响应的消息
channel.basicConsume(callbackQueueName, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
if (collId.equals(properties.getCorrelationId())) {
System.out.println(format("client recv message:{0}, corrId:{1}", new String(body), collId));
} else {
System.out.println("不是本次请求的消息");
}
}
});
TimeUnit.SECONDS.sleep(1);
RabbitMqUtils.close(connection, channel);
}
private static AMQP.BasicProperties getBasicPublishProperties(String corrId, String callbackQueueName) {
return new AMQP.BasicProperties().builder()
.correlationId(corrId)
.replyTo(callbackQueueName).build();
}
}
持久化：
在RabbitMQ中交换器、队列、消息都设置为持久化就能保持消息不丢失了嘛？
当然不，情况如下：
1、当autoAck设置为true的时候，消费者接收到消息后还没来得及处理就宕机了。
解决：autoAck设为false，消费者处理完消息后再通知服务端删除消息。
2、再RabbitMQ持久化到磁盘中的这段时间，RabbitMQ服务器宕机了。
解决：服务端确认机制、镜像队列（后面章节会描述）。
事务：
1、开启事务：channel.txSelect()
2、提交事务：channel.txCommit()
3、回滚事务：channel.txRollback()
事务和db的事务很相似，不细说。
发送方确认机制：
AMQP协议提供了事务机制来保证消息能真正成功的到达RabbitMQ，但事务机制会严重的影响到RabbitMQ的吞吐量，所以RabbitMQ引入了一种轻量的方式，发送方确认机制。
客户端使用方式：
1、将信道设置发送方确认方式：channel.confirmSelect()。
2、确认消息是否发送成功：
）boolean waitForConfirms() throws InterruptedException;
）boolean waitForConfirms(long timeout) throws InterruptedException, TimeoutException;
）void waitForConfirmsOrDie() throws IOException, InterruptedException;
）void waitForConfirmsOrDie(long timeout) throws IOException, InterruptedException, TimeoutException;
发送方确认消息成功的三种方式：
public class PublisherConfirmProduct {
private static final String EXCHANGE_NAME = "demo.exchange";
private static final String ROUTING_KEY = "demo.routingkey";
private static final String QUEUE_NAME = "demo.queue";
private static final String MESSAGE = "Hello World！";
/**
* 单条确认
*/
public static void commonConfirm() throws Exception {
Connection connection = RabbitMqUtils.getConnection();
Channel channel = initChannel(connection);
channel.confirmSelect();
for (int i = 0; i < 100; i++) {
channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, MessageProperties.PERSISTENT_TEXT_PLAIN, MESSAGE.getBytes());
if (channel.waitForConfirms()) {
// 逐条确认是否发送成功
System.out.println("send success!");
}
}
RabbitMqUtils.close(connection, channel);
}
/**
* 批量确认
*/
public static void batchConfirm() throws Exception {
Connection connection = RabbitMqUtils.getConnection();
Channel channel = initChannel(connection);
channel.confirmSelect();
for (int i = 0; i < 100; i++) {
channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, MessageProperties.PERSISTENT_TEXT_PLAIN, MESSAGE.getBytes());
}
// 批量确认是否发送成功，如果某一次确认失败这一批都要重新发送
if (channel.waitForConfirms()) {
System.out.println("send success!");
}
RabbitMqUtils.close(connection, channel);
}
/**
* 异步确认
*/
public static void asyncConfirm() throws Exception {
Connection connection = RabbitMqUtils.getConnection();
Channel channel = initChannel(connection);
channel.basicQos(1);
channel.confirmSelect();
// 定义一个未确认消息集合
final SortedSet<Long> unConfirmSet = Collections.synchronizedNavigableSet(new TreeSet<>());
for (int i = 0; i < 100; i++) {
channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, MessageProperties.PERSISTENT_TEXT_PLAIN, MESSAGE.getBytes());
unConfirmSet.add(channel.getNextPublishSeqNo());
}
channel.addConfirmListener(new ConfirmListener() {
@Override
public void handleNack(long deliveryTag, boolean multiple) throws IOException {
System.err.println(format("拒绝消息 deliveryTag:{0}, multiple:{1}", deliveryTag, multiple));
}
@Override
public void handleAck(long deliveryTag, boolean multiple) throws IOException {
System.err.println(format("确认消息 deliveryTag:{0}, multiple:{1}", deliveryTag, multiple));
if (multiple) {
// multiple为true，则deliveryTag之前的所有消息全部被确认
unConfirmSet.headSet(deliveryTag + 1).clear();
} else {
// 否则只确认一条消息
unConfirmSet.remove(deliveryTag);
}
}
});
TimeUnit.SECONDS.sleep(5);
System.out.println(unConfirmSet.size());
RabbitMqUtils.close(connection, channel);
}
private static Channel initChannel(Connection connection) throws IOException {
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT, true, false, null);
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
return channel;
}
public static void main(String[] args) throws Exception {
// commonConfirm();
// batchConfirm();
asyncConfirm();
}
}
最新文章
- Web Service 的创建简单编码、发布和部署
- WEB开发入门
- 使用Spring注解来简化ssh框架的代码编写
- 洛谷⑨月月赛Round2 P3393逃离僵尸岛[最短路]
- nanosleep() -- 更精确的延迟 -----一个使用用例
- git的基本使用
- tomcat 下虚拟机部署导致应用filter失效的问题
- 实用js+css多级树形展开效果导航菜单
- Java中各种排序算法
- Java编程陷阱-类成员初始化
- vs2005 测试 lua环境
- Maven+Spring
- 第21章 策略模式（Strategy Pattern）
- vue+webpack项目实际工作中需要生成一个配置文件供生产环境使用
- CSS中2d转换：transition过渡放在:hover伪类中与应用在整个元素中区别
- Centos7下,简单DOCKER 使用.映射SSH端口到宿主主机.
- javascript基础知识学习
- BZOJ1785[USACO 2010 Jan Gold 3.Cow Telephones]——贪心
- go语言学习--指针数组和数组指针
- Centos重新启动网络配置文件，/etc/resolv.conf被覆盖或清空问题解决