1、工作队列的概念
简单队列不足 : 不支持多个消费者；
工作队列即一个生产者可以对应多个消费者同时消费；
相比简单队列支持多消费者； 因为实际工作中,生产者服务一般都是很简单的业务逻辑处理之后就发送到队列,消费者接收到队列的消息之后,进行复杂的业务逻辑处理,所以一般都是多个消费者进行处理.如是是一个消费者进行处理,那么队列会积压很多消息.
2、轮循分发（round-robin）
在默认情况下, RabbitMQ将逐个发送消息到在序列中的下一个消费者(而不考虑每个任务处理的时长等等,且是提前一次性分配,并非一个一个的分配) . 平均每个消费者获取相同数量的消息. 这种分发消息机制称为 轮询分发
当消息进入队列 ,RabbitMQ就会分发消息 .它不看消费者的应答的数目 ,也不关心消费者处理消息的能力,只是盲目的将第n条消息发给第n个消费者
- 生产者
/**
* @author zhaod
* @description
* @date 2018/9/27 11:11
*/
public class Producer {
private static final Logger log = LoggerFactory.getLogger(Producer.class);
private static final String QUEUE_NAME = "my-work-queue";
public static void main(String[] args) throws IOException, InterruptedException, TimeoutException {
// 获取连接
Connection connection = MqConnectionUtil.getConnection();
// 创建信道
Channel channel = connection.createChannel();
// 申明队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
for (int i = 1; i < 11; i++) {
String msg = "I am " + i + " old";
channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
System.out.println("P---->" + msg);
Thread.sleep(500);
}
channel.close();
connection.close();
}
}
- 消费者1
/**
* @author zhaodi
* @description 工作队列
* @date 2018/9/27 11:30
*/
public class Consumer01 {
private static final Logger log = LoggerFactory.getLogger(Consumer01.class);
private static final String QUEUE_NAME = "my-work-queue";
public static void main(String[] args) throws IOException {
// 连接
Connection connection = MqConnectionUtil.getConnection();
// 信道
Channel channel = connection.createChannel();
// 队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 消费者
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String msg = "C-->{1}接收:" +new String(body );
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(msg);
}
};
// 监听
channel.basicConsume(QUEUE_NAME,true,consumer);
}
}
- 消费者2
/**
* @author zhaodi
* @description 工作队列
* @date 2018/9/27 11:30
*/
public class Consumer02 {
private static final Logger log = LoggerFactory.getLogger(Consumer02.class);
private static final String QUEUE_NAME = "my-work-queue";
public static void main(String[] args) throws IOException {
// 连接
Connection connection = MqConnectionUtil.getConnection();
// 信道
Channel channel = connection.createChannel();
// 队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 消费者
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String msg = "C-->{2}接收:" +new String(body);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(msg);
}
};
// 监听
channel.basicConsume(QUEUE_NAME,true,consumer);
}
}
3、公平分发（fair dispatch）
根据消费者处理性能,性能好的消费的数据量多,性能差的消费的数据量少 .这种分发消息机制称为 公平分发
如何实现公平分发
-
限制发给消费者的消息只可以有1条，在这个消费者确认消息之前，不可以发送吓一跳消息给这个消费者
int prefetch = 1; channel.basicQos(prefetch); -
默认自动应答改为手动应答
// 关闭自动应答 Boolean autoAck = false; // 监听 channel.basicConsume(QUEUE_NAME, autoAck, consumer); // 开启手动应答 channel.basicAck(envelope.getDeliveryTag(), false); -
生产者
/**
* @author zhaod
* @description 公平分发
* @date 2018/9/27 11:11
*/
public class Producer03 {
private static final Logger log = LoggerFactory.getLogger(Producer03.class);
private static final String QUEUE_NAME = "my-work-queue";
public static void main(String[] args) throws IOException, InterruptedException, TimeoutException {
// 获取连接
Connection connection = MqConnectionUtil.getConnection();
// 创建信道
Channel channel = connection.createChannel();
// 申明队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 每个消费者在发送确认消息之前，消息队列不发送下一个消息给该消费者，保证每次只处理一条消息
int prefetch = 1;
channel.basicQos(prefetch);
for (int i = 1; i < 20; i++) {
String msg = "I am " + i + " old";
channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
System.out.println("P---->" + msg);
Thread.sleep(500);
}
channel.close();
connection.close();
}
}
- 消费者1
/**
* @author zhaodi
* @description 工作队列公平分发
* @date 2018/9/27 11:30
*/
public class Consumer03 {
private static final Logger log = LoggerFactory.getLogger(Consumer03.class);
private static final String QUEUE_NAME = "my-work-queue";
public static void main(String[] args) throws IOException {
// 连接
Connection connection = MqConnectionUtil.getConnection();
// 信道
Channel channel = connection.createChannel();
// 队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
//
int prefetch = 1;
channel.basicQos(prefetch);
// 消费者
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String msg = "C-->{3}接收:" +new String(body );
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(msg);
// 处理完消息之后，发送回执，告诉队列，你给我的消息我处理完了，你可以发送下一条消息给我了
// envelope.getDeliveryTag(),标出是哪条消息
channel.basicAck(envelope.getDeliveryTag(),false);
}
};
// 自动应答关闭
boolean autoAck = false;
// 监听
channel.basicConsume(QUEUE_NAME,autoAck,consumer);
}
}
- 消费者2
代码同上面
4、消息应答机制
4.1 为什么需要消息应答机制
完成一个任务需要花费几秒钟，但是如果某个消费者开始执行某个任务花费了很长的时间并且在执行到某个部分的时候崩溃了怎么办。
在我们目前的代码中，在向消费者推送了某一条消息后，RabiitMQ会立即删除这条消息的。
如果我们kill掉某个worker的话，那么我们将会丢失该worker正在处理的消息，我们也会丢失掉所有被发送到这个消费者且未被处理完成的消息。
4.2 消息应答机制的作用
为了保证消息永远不会被丢失，RabbitMQ采用消息应答机制。
当消费者接收到消息并完成任务后会往RabbitMQ服务器发送一条确认的命令，然后RabbitMQ才会将消息删除。
如果某个消费者在还有发送确认信息就挂了，RabbitMQ将会视为服务没有执行完成，然后把执行消息的服务再发给另外一个消费者。这种方式下，即时某个worker挂了，也不会使得消息丢失。
这里不是用超时来判断的，只有在某个消费者连接断开时，RabbitMQ才会把重新发送该消费者没有返回确认的消息到其它消费者那。即时处理某条任务花费了很长的时间，在这里也是没有问题的。
消息应答机制默认是开启的，也就是说当消费者接收到消息的时候，不管是否开始处理接收到的消息，它已经向RabbitMQ发送确认消息，这时候RabbitMQ服务器就会删除该条消息。
// 自动应答关闭
boolean autoAck = false;
// 监听
channel.basicConsume(QUEUE_NAME,autoAck,consumer);
很多人都会忘记调用basicAck方法，虽然这是一个很简单的错误，但往往却是致命。消费者退出后消息将会被重发，但是由于一些未能被确认消息不能被释放，RabbitMQ将会消耗掉越来越多的内存
channel.basicAck(envelope.getDeliveryTag(),false);
5、消息持久化
如果RabbitMQ的服务器宕机那么怎么保证消息不丢失呢？
当MQ重启，那么之前的队列消息是会丢失的；
解决：将队列和消息都持久化存储
注意点：队列持久化的时候，生产者和消费者都要申明
// 申明队列，并且指明该队列是持久化的
Boolean durable = true;
channel.queueDeclare(QUEUE_NAME, durable, false, false, null);
...
// 发送消息，并且持久化消息
channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes());
消息的持久化不能百分百的保证消息不会丢失，虽然RabbitMQ会把消息写到磁盘上，但是从RabbitMQ接收到消息到写到磁盘上，这个短时间的过程中发生的RabbitMQ重启依然会使得为写入到磁盘的消息被丢失；
事实上是这样的，RabbitMQ接收到消息后，首先会把该消息写到内存缓冲区中，并不是直接把单条消息实时写到磁盘上的。消息的持久化不是健壮的，但是对于简单的任务队列是够用了。如果你需要一套很健壮的持久化方案，那么你可以使用publisher confirms