一、Hello World
1. 消费者发送单个消息的生产者 和 接收消息并打印出来的消费者：
在图中，“P” 是我们的生产者，“C” 是我们的消费者。中间的框是一个队列 RabbitMQ 代表使用者保留的消息缓冲区。
public class Consumer {
private static final String QUEUE_NAME = "hello";
public static void main(String[] args) {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("192.168.137.155");
factory.setUsername("guest");
factory.setPassword("guest");
try {
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.basicConsume(QUEUE_NAME, true,
// 接收消息
(consumerTag, message) -> {
System.out.println(message);
System.out.printf("接收到消息: %srn", new String(message.getBody()));
},
// 取消消息
consumerTag -> {
System.out.printf("中断消费消息: %srn", consumerTag);
});
System.out.println("等待接收消费...");
} catch (IOException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
}
}
}
2. 生产者
public class Producer {
// 队列名称
private static final String QUEUE_NAME = "hello";
public static void main(String[] args) {
// 创建一个连接工厂
ConnectionFactory factory = new ConnectionFactory();
// 工厂IP，连接RabbitMQ的队列
factory.setHost("192.168.137.155");
// 用户名
factory.setUsername("guest");
// 密码
factory.setPassword("guest");
try {
// 创建连接
Connection connection = factory.newConnection();
// 获取信道
Channel channel = connection.createChannel();
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 消息
String message = "Hello World!";
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println("消息发送完毕");
} catch (IOException | TimeoutException e) {
e.printStackTrace();
}
}
}
二、工作队列(Work Queues)
1. 消费者工作队列(又称任务队列) 的主要思想是，避免立即执行资源密集型任务，而不得不等待它完成。相反我们安排任务在之后执行。把任务封装为消息并将其发送到队列。在后台运行的工作线程，将d出的任务最终执行作业。当有多个工作线程时，这些工作线程将一起处理这些任务。
public class Consumer {
private static final String QUEUE_NAME = "hello";
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = RabbitMQUtil.getChannel();
// 消费者消费消息
channel.basicConsume(QUEUE_NAME, true,
// 接收消息
(consumerTag, message) -> {
System.out.printf("接收到消息: %srn", new String(message.getBody()));
},
// 取消消息
consumerTag -> {
System.out.printf("中断消费消息: %srn", consumerTag);
});
System.out.println("Consumer, 2，等待接收消息...");
}
}
2. 生产者
public class Producer {
private static final String QUEUE_NAME = "hello";
public static void main(String[] args) throws IOException, TimeoutException {
// 获取信道
Channel channel = RabbitMQUtil.getChannel();
// 声明一个队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 从控制台接收消息
Scanner scanner = new Scanner(System.in);
System.out.println("输入消息: ");
while (scanner.hasNext()) {
String message = scanner.next();
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.printf("发送消息: %srn", message);
}
}
}
3. 启动两个消费者，轮训分发(prefetchCount = 0)
4. 不公平分发(prefetchCount = 1)在这个案例中我们会启动两个工作线程，一个消息发送线程，我们来看看他们两个工作线程是如何工作的。
在最开始的时候我们知道 RabbitMQ 分发消息采用的 轮训分发，但是在某种场景下这种策略并不是很好，比方说有两个消费者在处理任务，其中有个 消费者1 处理任务的速度非常快，而另外一个 消费者2 处理任务速度却很慢，这个时候我们还是采用 轮训分发 的话，就会导致处理速度快的这个消费者很大一部分时间处于空闲状态，而处理速度慢的那个消费者却一直在干活，这种分配方式在这种情况下其实就不太好，但是 RabbitMQ 并不知道这种情况它依然很公平的进行分发。
int prefetchCount = 1; // int prefetchCount = 2; channel.basicQos(prefetchCount);
6. 预取值(prefetchCount = 预取值）意思就是如果这个任务我还没有处理完，或者我还没有应答你，你先别分配给我新任务，我目前只能处理一个任务，然后 RabbitMQ 就会把新任务分配给其他空闲的消费者，当然如果所有的消费者都没有完成手上任务，队列还在不停的添加新任务，队列有可能就会遇到队列被撑满的情况，这个时候就只能添加新的 Worker 或者改变其他存储任务的策略。
本身消息就是异步发送的，所以在任何时候，Channel 上肯定不止只有一个消息，另外来自消费者的手动确认本质上也是异步的。因此这里就存在一个未确认的消息缓冲区，因此希望开发人员能 限制此缓冲区的大小，以避免缓冲区无限制添加未确认消息的问题。这个时候就可以通过使用 basicQos 方法设置 预取计数 值来完成的。该值定义通道上允许的未确认消息的最大数量。一旦数量达到配置的数量，RabbitMQ 将停止在通道上传递更多消息，除非有未处理的消息被确认。
例如，假设在通道上有未确认的消息 5、6、7，8，并且通道的预取计数设置为 4，此时 RabbitMQ 将不会在该通道上再传递任何消息，除非至少有一个未应答的消息被 ACK。比方说 tag=6 这个消息刚刚被确认 ACK，RabbitMQ 将会感知这个情况到并再发送一条消息。
消息应答 和 Qos预取值 对用户吞吐量有重大影响。通常增加预取值将提高向消费者传递消息的速度。虽然自动应答传输消息速率是最佳的，但是，在这种情况下已传递但尚未处理的消息数量也会增加，从而增加了消费者的 RAM 消耗(随机存取存储器)，应该小心使用具有 无限预处理的自动确认模式 或 手动确认模式，消费者消费了大量的消息如果没有确认的话，会导致消费者连接节点的内存消耗变大，所以找到合适的预取值是一个反复试验的过程，不同的负载该值取值也不同 100 到 300 范围内的值通常可提供最佳的吞吐量，并且不会给消费者带来太大的风险。预取值为 1 是最保守的。当然这将使吞吐量变得很低，特别是消费者连接延迟很严重的情况下，在消费者连接等待时间较长的环境中。对于大多数应用来说，稍微高一点的值将是最佳的。
三、发布订阅(Fanout交换机)
在上一节中，我们创建了一个工作队列。假设的是工作队列背后，每个任务都恰好交付给一个消费者(工作进程)。在这一部分中，我们将做一些完全不同的事情，将消息传达给多个消费者。这种模式
称为 发布订阅.
为了说明这种模式，我们将构建一个简单的日志系统。它将由两个程序组成：第一个程序将发出日志消息，第二个程序是消费者。
其中我们会启动两个消费者，其中一个消费者接收到消息后把日志存储在磁盘，另外一个消费者接收到消息后把日志打印在控制台上，事实上第一个程序发出的消息将广播给所有消费者。
四、路由(Direct交换机)
五、主题(Topic交换机)
六、发布确认 1. 发布确认原理
生产者将信道设置成 Confirm 模式，一旦信道进入 Confirm 模式，*所有在该信道上面发布的消息都将会被指派一个唯一的 ID(从 1 开始)，一旦消息被投递到所有匹配的队列之后，Broker 就会发送一个确认给生产者(包含消息的唯一 ID)，这就使得生产者知道消息已经正确到达目的队列了，如果消息和队列是可持久化的，那么确认消息会在将消息写入磁盘之后发出，Broker 回传给生产者的确认消息中 deliveryTag 域包含了确认消息的序列号，此外 Broker 也可以设置 basicAck 的 multiple 域，表示到这个序列号之前的所有消息都已经得到了处理。
2. 开启发布确认的方法Confirm 模式 最大的好处在于他是异步的，一旦发布一条消息，生产者应用程序就可以在等信道返回确认的同时，继续发送下一条消息，当消息最终得到确认之后，生产者应用便可以通过回调方法来处理该确认消息，如果 RabbitMQ 因为自身内部错误导致消息丢失，就会发送一条 nack 消息，生产者应用程序同样可以在回调方法中处理该 nack 消息。
发布确认默认是没有开启的，如果要开启需要调用方法 /confirm/iSelect，每当你要想使用发布确认，都需要在 Channel 上调用该方法
// 开启发布确认 channel./confirm/iSelect();3. 发布确认策略 3.1 单个确认发布
这是一种简单的确认方式，它是一种 同步确认发布 的方式，也就是发布一个消息之后只有它被确认发布，后续的消息才能继续发布，waitForConfirmsOrDie(long) 这个方法只有在消息被确认的时候才返回，如果在指定时间范围内这个消息没有被确认。那么它将抛出异常。
这种确认方式有一个最大的缺点就是：发布速度特别的慢，因为如果没有确认发布的消息就会阻塞所有后续消息的发布，这种方式最多提供每秒不超过数百条发布消息的吞吐量。当然对于某些应用程序来说这可能已经足够了。
private static void publishMessageIndividually() throws IOException, TimeoutException, InterruptedException {
Channel channel = RabbitMQUtil.getChannel();
// 队列声明
String queueName = UUID.randomUUID().toString();
channel.queueDeclare(queueName, true, false, false, null);
// 开启发布确认
channel./confirm/iSelect();
// 开始时间
long begin = System.currentTimeMillis();
for (int i = 0; i < MESSAGE_COUNT; i++) {
String message = String.format("单个确认发布: %d", i);
channel.basicPublish("", queueName, null, message.getBytes("UTF-8"));
// 单个消息确认
boolean flag = channel.waitFor/confirm/is();
if (flag) {
System.out.printf("%s, 确认rn", message);
}
}
System.out.printf("%d个消息，单个确认发布，耗时: %s毫秒", MESSAGE_COUNT, System.currentTimeMillis() - begin).println();
}
3.2 批量确认发布
上面那种方式非常慢，与单个等待确认消息相比，先发布一批消息然后一起确认可以极大地提高吞吐量，当然这种方式的缺点就是：当发生故障导致发布出现问题时，不知道是哪个消息出现问题了，我们必须将整个批处理保存在内存中，以记录重要的信息而后重新发布消息。当然这种方案仍然是同步的，也一样阻塞消息的发布。
private static void publishMessageBatch() throws IOException, TimeoutException, InterruptedException {
Channel channel = RabbitMQUtil.getChannel();
// 队列声明
String queueName = UUID.randomUUID().toString();
channel.queueDeclare(queueName, true, false, false, null);
// 开启发布确认
channel./confirm/iSelect();
// 批量确认消息数量
int batchSize = 100;
// 开始时间
long begin = System.currentTimeMillis();
for (int i = 0; i < MESSAGE_COUNT; i++) {
String message = String.format("批量确认发布: %d", i);
channel.basicPublish("", queueName, null, message.getBytes("UTF-8"));
// 判断发送消息达到100条，批量确认一次
if (i % batchSize == 0) {
channel.waitFor/confirm/is();
System.out.printf("%s, 确认rn", message);
}
}
System.out.printf("%d个消息，批量确认发布，耗时: %s毫秒", MESSAGE_COUNT, System.currentTimeMillis() - begin).println();
}
3.3 异步确认发布
异步确认虽然编程逻辑比上两个要复杂，但是性价比最高，无论是可靠性还是效率都没得说，他是利用回调函数来达到消息可靠性传递的，这个中间件也是通过函数回调来保证是否发送成功：
public static void publishMessageAsync() throws IOException, TimeoutException, InterruptedException {
Channel channel = RabbitMQUtil.getChannel();
// 队列声明
String queueName = UUID.randomUUID().toString();
channel.queueDeclare(queueName, true, false, false, null);
// 开启发布确认
channel./confirm/iSelect();
ConcurrentSkipListMap map = new ConcurrentSkipListMap<>();
// 消息确认监听器（异步通知）
channel.add/confirm/iListener(
// 监听那些消息成功
(deliveryTag, multiple) -> {
System.out.printf("成功的消息: %s", deliveryTag).println();
// 2. 删除已确认的消息
if (multiple) {
// 批量清除
ConcurrentNavigableMap concurrentNavigableMap = map.headMap(deliveryTag);
concurrentNavigableMap.clear();
} else {
map.remove(deliveryTag);
}
},
// 监听那些消息失败
(deliveryTag, multiple) -> {
System.out.printf("失败的消息: %s", deliveryTag).println();
System.out.println(map.get(deliveryTag));
});
// 开始时间
long begin = System.currentTimeMillis();
for (int i = 0; i < MESSAGE_COUNT; i++) {
String message = String.format("异步确认发布: %d", i);
channel.basicPublish("", queueName, null, message.getBytes("UTF-8"));
// 1. 记录所有发送的消息
map.put(channel.getNextPublishSeqNo(), message);
}
System.out.printf("%d个消息，异步确认发布，耗时: %s毫秒", MESSAGE_COUNT, System.currentTimeMillis() - begin).println();
}
3.3.1 处理异步未确认的消息
3.4 三种发布确认对比最好的解决的解决方案，就是把未确认的消息放到一个基于内存的能被发布线程访问的队列，比如说用 ConcurrentlinkedQueue 这个队列在 confirm callbacks 与发布线程之间进行消息的传递。
七、发布确认高级
1. 确认机制方案 2. 架构图 3 配置文件在生产环境中由于一些不明原因，导致 RabbitMQ 重启，在 RabbitMQ 重启期间生产者消息投递失败，导致消息丢失，需要手动处理和恢复。于是，我们开始思考，如何才能进行 RabbitMQ 的消息可靠投递呢？特别是在这样比较极端的情况，RabbitMQ 集群不可用的时候，无法投递的消息该如何处理呢？
# 发布消息成功到交换机后会触发回调方法 spring.rabbitmq.publisher-/confirm/i-type=correlated
@Configuration
public class /confirm/iBackupConfig {
// 确认交换机
public static final String /confirm/i_EXCHANGE_NAME = "/confirm/i.exchange";
// 确认队列
public static final String /confirm/i_QUEUE_NAME = "/confirm/i.queue";
// 确认RoutingKey
public static final String /confirm/i_ROUTING_KEY = "key1";
@Bean
public DirectExchange /confirm/iExchange() {
// return new DirectExchange(/confirm/i_EXCHANGE_NAME);
// 但指定了备份交换机就不返回了（备份交换机优先级高）
return ExchangeBuilder.directExchange(/confirm/i_EXCHANGE_NAME)
// 是否持久化
.durable(true)
// 指定备份交换机
.withArgument("alternate-exchange", BACKUP_EXCHANGE_NAME).build();
}
@Bean
public Queue /confirm/iQueue() {
return QueueBuilder.durable(/confirm/i_QUEUE_NAME).build();
}
@Bean
public Binding /confirm/iQueueBinding/confirm/iExchange(@Qualifier("/confirm/iQueue") Queue /confirm/iQueue,
@Qualifier("/confirm/iExchange") DirectExchange /confirm/iExchange) {
return BindingBuilder.bind(/confirm/iQueue).to(/confirm/iExchange).with(/confirm/i_ROUTING_KEY);
}
}
5. 消费者
@Slf4j
@Component
public class /confirm/iConsumer {
@RabbitListener(queues = /confirm/iBackupConfig./confirm/i_QUEUE_NAME)
public void receiveMsg(Message message) {
log.info("确认消费者接收到消息: {}", new String(message.getBody()));
}
}
6. 回调接口
@Slf4j
@Component
public class MyCallback implements RabbitTemplate./confirm/iCallback {
@Autowired
private RabbitTemplate rabbitTemplate;
@PostConstruct
public void init() {
// 将当前回调类注入RabbitTemplate
rabbitTemplate.set/confirm/iCallback(this);
}
@Override
public void /confirm/i(CorrelationData correlationData, boolean ack, String cause) {
String id = correlationData != null ? correlationData.getId() : "";
if (ack) {
log.info("交换机已收到Id为: {}的消息", id);
} else {
log.error("交换机未收到Id为: {}的消息，原因: {}", id, cause);
}
}
}
7. 生产者
@Slf4j
@RestController
@RequestMapping("//confirm/i")
public class ProduceController {
@Autowired
private RabbitTemplate rabbitTemplate;
@GetMapping("/sendMsg/{msg}")
public String sendMsg(@PathVariable String msg) {
log.info("发送消息: {}", msg);
CorrelationData correlationData = new CorrelationData("1");
// 1. 成功确认消息
// rabbitTemplate.convertAndSend(/confirm/iConfig./confirm/i_EXCHANGE_NAME, /confirm/iConfig./confirm/i_ROUTING_KEY, msg, correlationData);
// 2. 未知的交换机
// rabbitTemplate.convertAndSend(/confirm/iConfig./confirm/i_EXCHANGE_NAME + "123", /confirm/iConfig./confirm/i_ROUTING_KEY, msg, correlationData);
// 3. 未知的routingKey
rabbitTemplate.convertAndSend(/confirm/iBackupConfig./confirm/i_EXCHANGE_NAME, /confirm/iBackupConfig./confirm/i_ROUTING_KEY + "123", msg, correlationData);
return "success";
}
}
8. 回退消息 8.1 Mandatory参数可以看到，发送了两条消息，第一条消息的 RoutingKey 为 key1，第二条消息的 RoutingKey 为 key1 + 123，两条消息都成功被交换机接收，也收到了交换机的确认回调，但消费者只收到了一条消息，因为第二条消息的 RoutingKey 与队列的 BindingKey 不一致，也没有其它队列能接收这个消息，所有第二条消息被直接丢弃了。
9. 回调接口在仅开启了生产者确认机制的情况下，交换机接收到消息后，会直接给消息生产者发送确认消息，如果发现该消息不可路由，那么消息会被直接丢弃，此时生产者是不知道消息被丢弃这个事件的。那么如何让无法被路由的消息，帮我想办法处理一下？最起码通知我一声，我好自己处理啊。通过设置 mandatory 参数可以在当消息传递过程中，不可达目的地时将消息返回给生产者。
@Slf4j
@Component
public class MyCallback implements RabbitTemplate./confirm/iCallback, RabbitTemplate.ReturnCallback {
@Autowired
private RabbitTemplate rabbitTemplate;
@PostConstruct
public void init() {
// 将当前回调类注入RabbitTemplate
rabbitTemplate.set/confirm/iCallback(this);
rabbitTemplate.setMandatory(true);
rabbitTemplate.setReturnCallback(this);
}
@Override
public void /confirm/i(CorrelationData correlationData, boolean ack, String cause) {
String id = correlationData != null ? correlationData.getId() : "";
if (ack) {
log.info("交换机已收到Id为: {}的消息", id);
} else {
log.error("交换机未收到Id为: {}的消息，原因: {}", id, cause);
}
}
@Override
public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
log.error("消息: {}, 被{}交换机退回, 退回原因: {}, RoutingKey: {}",
new String(message.getBody()), exchange, replyText, routingKey);
}
}
10. 备份交换机
有了 Mandatory参数 和 回退消息，我们获得了对无法投递消息的感知能力，有机会在生产者的消息无法被投递时发现并处理。但有时候，我们并不知道该如何处理这些无法路由的消息，最多打个日志，然后触发报警，再来手动处理。而通过日志来处理这些无法路由的消息是很不优雅的做法，特别是当生产者所在的服务有多台机器的时候，手动复制日志会更加麻烦而且容易出错。而且设置 mandatory 参数会增加生产者的复杂性，需要添加处理这些被退回的消息的逻辑。如果既不想丢失消息，又不想增加生产者的复杂性，该怎么做呢？
前面在设置死信队列的文章中，我们提到，可以为队列设置 死信交换机 来存储那些处理失败的消息，可是这些不可路由消息根本没有机会进入到队列，因此无法使用死信队列来保存消息。在 RabbitMQ 中，有一种 备份交换机 的机制存在，可以很好的应对这个问题。什么是备份交换机呢？
11. 备份交换机架构图备份交换机可以理解为 RabbitMQ 中 交换机的备胎，当我们为某一个交换机声明一个对应的备份交换机时，就是为它创建一个备胎，当交换机接收到一条不可路由消息时，将会把这条消息转发到备份交换机中，由备份交换机来进行转发和处理，通常 备份交换机 的类型为 Fanout ，这样就能把所有消息都投递到与其绑定的队列中，然后我们在备份交换机下绑定一个队列，这样所有那些原交换机无法被路由的消息，就会都进入这个队列了。当然，我们还可以建立一个报警队列，用独立的消费者来进行监测和报警。
@Configuration
public class /confirm/iBackupConfig {
// 确认交换机
public static final String /confirm/i_EXCHANGE_NAME = "/confirm/i.exchange";
// 确认队列
public static final String /confirm/i_QUEUE_NAME = "/confirm/i.queue";
// 确认RoutingKey
public static final String /confirm/i_ROUTING_KEY = "key1";
@Bean
public DirectExchange /confirm/iExchange() {
// return new DirectExchange(/confirm/i_EXCHANGE_NAME);
// 但指定了备份交换机就不返回了（备份交换机优先级高）
return ExchangeBuilder.directExchange(/confirm/i_EXCHANGE_NAME)
// 是否持久化
.durable(true)
// 指定备份交换机
.withArgument("alternate-exchange", BACKUP_EXCHANGE_NAME).build();
}
@Bean
public Queue /confirm/iQueue() {
return QueueBuilder.durable(/confirm/i_QUEUE_NAME).build();
}
@Bean
public Binding /confirm/iQueueBinding/confirm/iExchange(@Qualifier("/confirm/iQueue") Queue /confirm/iQueue,
@Qualifier("/confirm/iExchange") DirectExchange /confirm/iExchange) {
return BindingBuilder.bind(/confirm/iQueue).to(/confirm/iExchange).with(/confirm/i_ROUTING_KEY);
}
// 备份交换机
public static final String BACKUP_EXCHANGE_NAME = "backup.exchange";
// 备份队列
public static final String BACKUP_QUEUE_NAME = "backup.queue";
// 报警队列
public static final String WARNING_QUEUE_NAME = "warning.queue";
@Bean
public FanoutExchange backupExchange() {
return new FanoutExchange(BACKUP_EXCHANGE_NAME);
}
@Bean
public Queue backupQueue() {
return QueueBuilder.durable(BACKUP_QUEUE_NAME).build();
}
@Bean
public Queue warningQueue() {
return QueueBuilder.durable(WARNING_QUEUE_NAME).build();
}
@Bean
public Binding backupQueueBindingBackupExchange(@Qualifier("backupQueue") Queue backupQueue,
@Qualifier("backupExchange") FanoutExchange backupExchange) {
return BindingBuilder.bind(backupQueue).to(backupExchange);
}
@Bean
public Binding warningQueueBindingBackupExchange(@Qualifier("warningQueue") Queue warningQueue,
@Qualifier("backupExchange") FanoutExchange backupExchange) {
return BindingBuilder.bind(warningQueue).to(backupExchange);
}
}
12. 报警消费者
@Slf4j
@Component
public class WarningConsumer {
@RabbitListener(queues = /confirm/iBackupConfig.WARNING_QUEUE_NAME)
public void receiveMsg(Message message) {
log.warn("报警消费者接收到消息: {}", new String(message.getBody()));
}
}
13. 测试注意事项
- Mandatory参数 与 备份交换机 可以一起使用的时候，如果两者同时开启，消息究竟何去何从？谁优先级高？
经过上面结果显示答案是 备份交换机优先级高。
八、其他 1. 消息应答 1.1 概念
消费者完成一个任务可能需要一段时间，如果其中一个消费者处理一个很长的任务，并且只完成了部分任务突然就挂掉了，会发生什么情况。RabbitMQ 一旦向消费者传递了一条消息，便立即将该消息标记为删除。在这种情况下，突然有个消费者挂掉了，我们将丢失正在处理的消息。以及后续再发送给该消费者的消息，它都无法接收到。
1.2 自动应答为了保证消息在发送过程中不丢失，RabbitMQ 引入了 消息应答机制，消息应答机制就是：消费者在接收到消息，并且处理该消息之后，告诉 RabbitMQ 它已经处理，RabbitMQ 可以把该消息删除了。
1.3 手动应答消息发送后立即被认为已经传送成功，这种模式需要在 高吞吐量 和 数据传输安全性方面 做权衡，因为这种模式如果消息在接收到之前，消费者那边出现 连接 或 Channel 关闭，那么消息就丢失了，当然另一方面这种模式消费者那边可以传递过载的消息，没有对传递的消息数量进行限制，当然这样有可能使得消费者这边由于接收太多还来不及处理的消息，导致这些消息的积压，最终使得内存耗尽，最终这些消费者线程被 *** 作系统杀死，所以这种模式仅适用在消费者可以高效并以某种速率能够处理这些消息的情况下使用。
1.3.1 手动应答的方法默认消息采用的是 自动应答，所以我们要想实现消息消费过程中不丢失，需要把 自动应答 改为 手动应答，修改消费者代码。
// 用于肯定确认（告诉 RabbitMQ 该消息成功的被处理，可以将其丢弃了） void basicAck(long deliveryTag, boolean multiple); // 用于否定确认 void basicNack(long deliveryTag, boolean multiple, boolean requeue); // 用于拒绝消息（不处理该消息了直接拒绝，可以将其丢弃了） void basicReject(long deliveryTag, boolean requeue);
1.3.2 消息自动重新入队手动应答的好处是可以批量应答并且减少网络拥堵：
- multiple 的 true 和 false 代表不同意思：
- true 代表批量应答 Channel 上未应答的消息
比如说 Channel 上有传送 tag 的消息 5, 6, 7, 8 当前 tag 是 8，那么此时 5-8 的这些还未应答的消息都会被批量确认收到消息应答；- false 同上面相比
只会应答 tag=8 的消息，5, 6, 7 这三个消息依然不会被确认收到消息应答；
1.2.3 消费者如果消费者由于某些原因失去连接(通道已关闭、连接已关闭 或 TCP 连接丢失)，导致消息未发送 ACK 确认，RabbitMQ 将了解到消息未完全处理，并将对其重新排队。如果此时其他消费者可以处理，它将很快将其重新分发给另一个消费者。这样，即使某个消费者偶尔死亡，也可以确保不会丢失任何消息。
// 1. false手动应答 boolean autoAck = false; channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
public class Consumer1 {
private static final String QUEUE_NAME = "ack_queue";
public static void main(String[] args) throws IOException, TimeoutException {
System.out.println("Consumer1，睡1秒");
// 获取信道
Channel channel = RabbitMQUtil.getChannel();
// int prefetchCount = 1;
int prefetchCount = 2;
channel.basicQos(prefetchCount);
// 1. false手动应答
boolean autoAck = false;
channel.basicConsume(QUEUE_NAME, autoAck,
// 接收消息
(consumerTag, message) -> {
try {
// 睡1秒
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.printf("接收到消息: %srn", new String(message.getBody(), "UTF-8"));
channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
},
// 取消消息
consumerTag -> {
System.out.printf("中断消费消息: %srn", consumerTag);
});
}
}
1.2.4 生产者
默认消息采用的是自动应答，所以我们要想实现消息消费过程中不丢失，需要把自动应答改为手动应答；
public class Producer {
private static final String QUEUE_NAME = "ack_queue";
public static void main(String[] args) throws IOException, TimeoutException {
// 获取信道
Channel channel = RabbitMQUtil.getChannel();
// 2. 队列持久化
boolean durable = true;
channel.queueDeclare(QUEUE_NAME, durable, false, false, null);
// 从控制台接收消息
Scanner scanner = new Scanner(System.in);
System.out.println("输入消息: ");
while (scanner.hasNext()) {
String message = scanner.next();
// 3. 消息持久化
AMQP.BasicProperties props = MessageProperties.PERSISTENT_TEXT_PLAIN;
channel.basicPublish("", QUEUE_NAME, props, message.getBytes("UTF-8"));
System.out.printf("发送消息: %srn", message);
}
}
}
2. RabbitMQ持久化
2.1 概念
2.2 队列持久化(Features = D)刚刚我们已经看到了如何 处理任务不丢失 的情况，但是如何保障当 RabbitMQ 服务停掉以后，消息生产者发送过来的 消息不丢失。默认情况下 RabbitMQ 退出或由于某种原因崩溃时，它忽视队列和消息，除非告知它不要这样做。确保消息不会丢失需要做两件事：我们需要将 队列 和 消息 都标记为持久化。
之前我们创建的队列都是非持久化的，RabbitMQ 如果重启的话，该队列就会被删除掉，如果要队列实现持久化，需要在声明队列的时候把 durable 参数设置为持久化。
// 2. 队列持久化 boolean durable = true; channel.queueDeclare(QUEUE_NAME, durable, false, false, null);
2.2 消息持久化以下为控制台中 持久化 与 非持久化 队列的 UI 显示区：
这个时候即使重启 RabbitMQ 队列也依然存在。
要想让消息实现持久化，需要在消息生产者端添加这个属性：MessageProperties.PERSISTENT_TEXT_PLAIN
// 3. 消息持久化
AMQP.BasicProperties props = MessageProperties.PERSISTENT_TEXT_PLAIN;
channel.basicPublish("", QUEUE_NAME, props, message.getBytes("UTF-8"));
将消息标记为持久化，并不能完全保证消息不会丢失。尽管它告诉 RabbitMQ 将消息保存到磁盘，但是这里依然存在当消息刚准备存储在磁盘的时候，但是还没有存储完，消息还在缓存的一个间隔点。此时并没有真正写入磁盘。持久性保证并不强，但是对于我们的简单任务队列而言，这已经绰绰有余了。如果需要更强有力的持久化策略，参考 发布确认。
欢迎分享，转载请注明来源：内存溢出
微信扫一扫
支付宝扫一扫
评论列表（0条）