- 1. Work Quesues概述
- 2. Work Queues案例
work queues是工作队列，又叫任务队列，是为了解决多个消费者有序执行密集型的资源任务。比如以下图为例，生产者产生大量消息发给了消息队列中，但是消息消费时比较耗时，这种情况下可以创建多个消费者同时进行消费，但一定要保证一个消息只能被一个消费者进行消费，假如一个消息被Consumer1消费了就不能再被Consumer2或者Consumer3再消费，所以queue队列中消息就像有序的被消费，queue队列中第一个消息分配给了Consumer1进行消费，第二个消息会被分配给Consumer2进行消费，第三个消息被分配给Consumer3进行消费，然后再依次进行消费……，这种现象就是work queues的轮询消费形式。
假如一个生产者连续生产了10个消息，3个消费者依次消费这10个消息。
首先创建一个连接RabbitMQ的工具类，生产者和消费者都需要通过这个工具类与RabbitMQ服务器进行连接。
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class RabbitmqUtil {
private final static String HOST_NM = "192.168.85.100";
private final static String USER_NM = "admin";
private final static String PASSWORD = "123";
public static Connection getConnection() throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost(HOST_NM);
factory.setUsername(USER_NM);
factory.setPassword(PASSWORD);
Connection connection = factory.newConnection();
return connection;
}
public static Channel getChannel() throws IOException, TimeoutException {
Connection connection = RabbitmqUtil.getConnection();
Channel channel = connection.createChannel();
return channel;
}
}
接下来创建一个生产者Producer，生产者连续生产10个消息存入消息队列。
import com.lzj.rabbitmq.RabbitmqUtil;
import com.rabbitmq.client.Channel;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Producer {
private final static String QUEUE_NAME = "hello_queue";
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = RabbitmqUtil.getChannel();
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
String message = null;
for(int i=0; i<10; i++){
message = "hello world " + i;
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println("hello world " + i + "消息发送完毕");
}
}
}
然后分别创建3个消费者Consumer1、Consumer2、Consumer3，3个消费者的代码一致，比如下面是Consumer1消费者的代码，然后复制2份分别改成Consumer2和Consumer3即可。
import com.lzj.rabbitmq.RabbitmqUtil;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Consumer1 {
private final static String QUEUE_NAME = "hello_queue";
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = RabbitmqUtil.getChannel();
DeliverCallback deliverCallback = (consumerTag, message) -> {
System.out.println("Consumer1消费的消息体为：" + new String(message.getBody()));
};
CancelCallback cancelCallback = consumerTag -> {
System.out.println("Consumer1取消消费消息的标签为：" + consumerTag);
};
channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);
}
}
通过上面几步，生产者和消费者都已创建好，下面验证一下测试结果，首先分别启动3个消费者程序，由于此时hello_queue队列中还没有消息，所以3个消费者都没有消息进行消费处于等待中。然后启动Producer程序，连续生产了10个消息，如下所示
hello world 0消息发送完毕 hello world 1消息发送完毕 hello world 2消息发送完毕 hello world 3消息发送完毕 hello world 4消息发送完毕 hello world 5消息发送完毕 hello world 6消息发送完毕 hello world 7消息发送完毕 hello world 8消息发送完毕 hello world 9消息发送完毕
然后查看Consumer1消费消息情况，发现Consumer1的终端输出如下所示，表示消费了下面4个消息。
Consumer1消费的消息体为：hello world 0 Consumer1消费的消息体为：hello world 3 Consumer1消费的消息体为：hello world 6 Consumer1消费的消息体为：hello world 9
Consumer2消费了下面3个消息
Consumer2消费的消息体为：hello world 1 Consumer2消费的消息体为：hello world 4 Consumer2消费的消息体为：hello world 7
Consumer3消费了下面3个消息
Consumer3消费的消息体为：hello world 2 Consumer3消费的消息体为：hello world 5 Consumer3消费的消息体为：hello world 8
3个消费者消费的消息总数正好等于生产者生产消息的总数，3个消费者分别依次消费hello_queue队列中的消息，保证了消息轮询被消费，没有被重复消费。
