一 ：Work Queues
Work Queues— 工作队列 (又称任务队列) 的主要思想是避免立即执行资源密集型任务，而不得不等待它完成。我们把任务封装为消息并将其发送到队列，在后台运行的工作进程将弹出任务并最终执行作业。当有多个工作线程时，这些工作线程将一起处理这些任务。
轮训分发消息
在这个案例中我们会启动两个工作线程，一个消息发送线程，我们来看看他们两个工作线程是如何工作的。
package com.oddfar.utils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class RabbitMqUtils {
//得到一个连接的 channel
public static Channel getChannel() throws Exception {
//创建一个连接工厂
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("42.192.149.71");
factory.setUsername("admin");
factory.setPassword("123456");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
return channel;
}
}
package com.oddfar.two;
import com.oddfar.utils.RabbitMqUtils;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
/**
* 这是一个工作线程，相当于之前的消费者
*
* @author zhiyuan
*/
public class Worker01 {
private static final String QUEUE_NAME = "hello";
public static void main(String[] args) throws Exception {
Channel channel = RabbitMqUtils.getChannel();
//消息接受
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String receivedMessage = new String(delivery.getBody());
System.out.println("接收到消息:" + receivedMessage);
};
//消息被取消
CancelCallback cancelCallback = (consumerTag) -> {
System.out.println(consumerTag + "消费者取消消费接口回调逻辑");
};
System.out.println("C1 消费者启动等待消费.................. ");
channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);
}
}
public class Task01 {
public static final String QUEUE_NAME = "hello";
public static void main(String[] args) throws Exception {
Channel channel = RabbitMqUtils.getChannel();
Scanner scanner = new Scanner(System.in);
while (scanner.hasNext()) {
String message = scanner.next();
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println("消息发送完成：" + message);
}
}
}
通过程序执行发现生产者总共发送 4 个消息，消费者 1 和消费者 2 分别分得两个消息，并且是按照有序的一个接收一次消息。这就是轮训的效果
另外：轮询分发并不是串行的轮询分发，假设生产者连续发1 - 10 这10个消息，消费者A会消费1,3,5,7,9，消费者2会消费到2,4,6,8,10 。消费者1比消费者2的消费速度快很多，当消费者1消费完1的时候，消费者2还在消费2,此时消费者1并不会等待消费者1完成消息2，消费者1会快速的将3,5,7,9这4个消息快速消费完，然后消费者2再慢慢的消费4,6,8,10。
此时消费者1它并不会去救济消费者2 并不会去帮消费者2消费消息
，也就是说它不会向forkJoin这种东西会动态的去执行任务。
- 按照消费者消费能力分发（在消费者代码中声明）
//设置权重分发 (消费者速度快的机器会消费到更多的消息，消费速度慢的机器会消费到更少的消息，就是能者多劳)
channel.basicQos(1);
- 权重分发
也可以按照机器的能力进行分发，和nginx权重类似
二 消息应答 (ack)
消费者完成一个任务可能需要一段时间，如果其中一个消费者处理一个长的任务并仅只完成了部分突然它挂掉了，会发生什么情况。
RabbitMQ 一旦向消费者传递了一条消息，便立即将该消息标记为删除 (自动ack的情况)。在这种情况下，突然有个消费者挂掉了，我们将丢失正在处理的消息，以及后续发送给该消费者的消息，因为它无法接收到。
为了保证消息在发送过程中不丢失，引入消息应答机制，消息应答就是：消费者在接收到消息并且处理该消息之后，告诉 rabbitmq 它已经处理了，rabbitmq 可以把该消息删除了。(这就是手动ack的情况)
public class Task02 {
private static final String TASK_QUEUE_NAME = "ack_queue";
public static void main(String[] args) throws Exception {
Channel channel = RabbitMqUtils.getChannel();
//声明队列
channel.queueDeclare(TASK_QUEUE_NAME, false, false, false, null);
Scanner sc = new Scanner(System.in);
System.out.println("请输入信息");
while (sc.hasNext()) {
String message = sc.nextLine();
//发布消息
channel.basicPublish("", TASK_QUEUE_NAME, null, message.getBytes("UTF-8"));
System.out.println("生产者发出消息" + message);
}
}
}
public class Work03 {
private static final String TASK_QUEUE_NAME = "ack_queue";
public static void main(String[] args) throws Exception {
Channel channel = RabbitMqUtils.getChannel();
System.out.println("C1 等待接收消息处理时间较 短");
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody());
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("接收到消息:" + message);
//todo 三个处理消息方法的api
/**
* 1.消息标记 tag
* 2.是否批量应答未应答消息
*/
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
/**
* 第一个参数：deliveryTag 消息的标识
* 第二个参数 multiple 是否批量确认消息
* 第三个参数：requeue 被Nack后 是否重新入队，
* 如果是填false 那么就相当于消息被消费者自己丢弃了，如果是true 那么相当于重新入队，broker会交给其他的消费者来消费这个消息
* 如果消费者仅有一个 然后Nack后的requeue参数又是true 那么就相当于中国消息一直会发送给这个消费者了，相当于消费者死循环的消费这个消息
*/
// channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false,false);
// basicNack和basicReject的区别仅仅是一个可以批量一个不能批量操作而已
// channel.basicReject();
};
CancelCallback cancelCallback = (s) -> {
System.out.println(s + "消费者取消消费接口回调逻辑");
};
//采用手动应答
boolean autoAck = false;
channel.basicConsume(TASK_QUEUE_NAME, autoAck, deliverCallback, cancelCallback);
}
}
public class Work04 {
private static final String TASK_QUEUE_NAME = "ack_queue";
public static void main(String[] args) throws Exception {
Channel channel = RabbitMqUtils.getChannel();
System.out.println("C1 等待接收消息处理时间较 长");
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody());
try {
Thread.sleep(10 * 1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("接收到消息:" + message);
/**
* 1.消息标记 tag
* 2.是否批量应答未应答消息
*/
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
};
CancelCallback cancelCallback = (s) -> {
System.out.println(s + "消费者取消消费接口回调逻辑");
};
//采用手动应答
boolean autoAck = false;
channel.basicConsume(TASK_QUEUE_NAME, autoAck, deliverCallback, cancelCallback);
}
}
最终效果，如果我们演示了消费消息较慢的消费者(Work04 )消费消息的时候 将程序断掉，那么消息相当于没有给broker确认ack，那么broker会将本应该属于Work04 的消息重新投递给其他消费者
另外：Nack和Reject的两种情况在Work03中写出来了
- 三 持久化
当 RabbitMQ 服务停掉以后，默认情况下会将队列和消息都会删除，
因此 ，我们在生产者发消息的时候
1：要将队列持久化，2： 要将消息进行持久化
public static void main(String[] args) throws Exception {
Channel channel = RabbitMqUtils.getChannel();
//1 设置队列持久化
boolean durable = true;
channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
Scanner sc = new Scanner(System.in);
System.out.println("请输入信息");
while (sc.hasNext()) {
String message = sc.nextLine();
//2 设置消息持久化
//发布消息
channel.basicPublish("", TASK_QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN , message.getBytes("UTF-8"));
System.out.println("生产者发出消息" + message);
}
}
网友评论