第四章-RabbitMQ工作模式-WORK
1.模式介绍
1.1 work模式
Work模式与前面的Simple模式一致，也是消息经由生产者发到Exchange再到queue再被消费者消费。不同点在于SIMPL模式是一个队列对应的一个消费者，此模式会由一个队列对应两个消费者或大于两个消费者。
1.2 work模式模拟
此模式下，我们的消费者实现了对消息的平均消费，但如果消费者1消费能力若于消费者2，那就会造成消费者1 的消息积压，这时候我们就会想到使用公平模式，能者多劳，消费快的多消费，消费少的少消费。合理消息。后面代码会涉及到。
2.验证代码
2.1 work平均模式
我们还是举小明洗澡的例子，小明洗澡比较孤独，这时候小明的女朋友小丽也来一块洗澡了，但两个人不想公用一个喷头，便又接入了一个喷头，一人一个开始洗，但是小明洗澡比较快，虽然洗完了，但是喷头关不上，只好把水浪费掉，小丽洗澡比较慢，但感觉水又不太够用。一共就20升水，一人10升。
2.1.1 生产者
/**
* @author rabbit
* @version 1.0.0
* @Description 一个生产者，一个默认的交换机，一个队列，两个消费者
* @createTime 2022/07/27 19:34:00
*/
public class WaterProducer {
public static final String QUEUE_NAME = "SolarWaterHeater";
//生产者
public static void main(String[] args) throws Exception {
//1、获取connection
Connection connection = RabbitCommonConfig.getConnection();
//2、创建channel
Channel channel = connection.createChannel();
for (int i = 1; i <= 20; i++) {
sendMsg(channel, i);
Thread.sleep(200);
}
//4、关闭管道和连接
channel.close();
connection.close();
}
private static void sendMsg(Channel channel, int k) throws IOException {
//3、发送消息到exchange
String msg = k + "升";
/**
* 参数1：指定exchange，使用“”。默认的exchange
* 参数2：指定路由的规则，使用具体的队列名称。exchange为""时，消息直接发送到队列中
* 参数3：指定传递的消息携带的properties
* 参数4：指定传递的消息，byte[]类型
*/
channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
System.out.println("水龙头放水成功！" + k + "升");
}
}2.1.2 消费者
小明洗澡：
/**
* @author rabbit
* @version 1.0.0
* @Description 工作模式 一个生产者，一个默认的交换机，一个队列，两个消费者
* 需要在consumer消费者端，平均分配
* @createTime 2022/07/27 19:36:00
*/
public class XMShowerConsumer {
public static final String QUEUE_NAME = "SolarWaterHeater";
//消费者
public static void main(String[] args) throws Exception {
//1、获取连对象、
Connection connection = RabbitCommonConfig.getConnection();
//2、创建channel
Channel channel = connection.createChannel();
//3、创建队列-helloworld
/**
* 参数1：queue 指定队列名称
* 参数2：durable 是否开启持久化（true）
* 参数3：exclusive 是否排外（当前队列只能被一个消费者消费）
* 参数4：autoDelete 如果这个队列没有其他消费者在消费，队列自动删除
* 参数5：arguments 指定队列携带的信息
*/
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
//4.开启监听Queue
DefaultConsumer consumer = new DefaultConsumer(channel) {
int i = 1;
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
try {
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
if (i > 5) {
System.out.println("小明洗澡洗完，开始浪费-第：" + new String(body, "UTF-8"));
} else {
System.out.println("小明洗澡已用水-第: " + new String(body, "UTF-8"));
}
i++;
}
};
/**
* 参数1：queue 指定消费哪个队列
* 参数1：deliverCallback 指定是否ACK（true:收到消息会立即告诉RabbiMQ，false：手动告诉）
* 参数1：cancelCallback 指定消费回调
*/
channel.basicConsume(QUEUE_NAME, true, consumer);
System.out.println("小明开始洗澡-快速洗......");
//5、键盘录入，让程序不结束！
System.in.read();
//6、释放资源
channel.close();
connection.close();
}
}小丽洗澡：
/**
* @author rabbit
* @version 1.0.0
* @Description 工作模式 一个生产者，一个默认的交换机，一个队列，两个消费者
* 需要在consumer消费者端，平均分配
* @createTime 2022/07/27 19:36:00
*/
public class XLShowerConsumer {
public static final String QUEUE_NAME = "SolarWaterHeater";
//消费者
public static void main(String[] args) throws Exception {
//1、获取连对象、
Connection connection = RabbitCommonConfig.getConnection();
//2、创建channel
Channel channel = connection.createChannel();
//3、创建队列-helloworld
/**
* 参数1：queue 指定队列名称
* 参数2：durable 是否开启持久化（true）
* 参数3：exclusive 是否排外（当前队列只能被一个消费者消费）
* 参数4：autoDelete 如果这个队列没有其他消费者在消费，队列自动删除
* 参数5：arguments 指定队列携带的信息
*/
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
//4.开启监听Queue
DefaultConsumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("小丽洗澡已用水-第: " + new String(body, "UTF-8"));
}
};
/**
* 参数1：queue 指定消费哪个队列
* 参数1：deliverCallback 指定是否ACK（true:收到消息会立即告诉RabbiMQ，false：手动告诉）
* 参数1：cancelCallback 指定消费回调
*/
channel.basicConsume(QUEUE_NAME, true, consumer);
System.out.println("小丽开始洗澡-慢慢洗......");
//5、键盘录入，让程序不结束！
System.in.read();
//6、释放资源
channel.close();
connection.close();
}
}
2.1.3 结果验证
生产者放水消息：
水龙头放水成功！1升
水龙头放水成功！2升
水龙头放水成功！3升
水龙头放水成功！4升
水龙头放水成功！5升
水龙头放水成功！6升
水龙头放水成功！7升
水龙头放水成功！8升
水龙头放水成功！9升
水龙头放水成功！10升
水龙头放水成功！11升
水龙头放水成功！12升
水龙头放水成功！13升
水龙头放水成功！14升
水龙头放水成功！15升
水龙头放水成功！16升
水龙头放水成功！17升
水龙头放水成功！18升
水龙头放水成功！19升
水龙头放水成功！20升小明洗澡
小明开始洗澡-快速洗......
小明洗澡已用水-第: 2升
小明洗澡已用水-第: 4升
小明洗澡已用水-第: 6升
小明洗澡已用水-第: 8升
小明洗澡已用水-第: 10升
小明洗澡洗完，开始浪费-第：12升
小明洗澡洗完，开始浪费-第：14升
小明洗澡洗完，开始浪费-第：16升
小明洗澡洗完，开始浪费-第：18升
小明洗澡洗完，开始浪费-第：20升小丽洗澡
小丽洗澡已用水-第: 1升
小丽洗澡已用水-第: 3升
小丽洗澡已用水-第: 5升
小丽洗澡已用水-第: 7升
小丽洗澡已用水-第: 9升
小丽洗澡已用水-第: 11升
小丽洗澡已用水-第: 13升
小丽洗澡已用水-第: 15升
小丽洗澡已用水-第: 17升
小丽洗澡已用水-第: 19升从结果来看，两个人都是平均用水，虽然有快有慢，慢的就会存在堆积情况。
2.2 公平模式
小丽就有意见了，自己不够洗，小明还白白浪费那么多水。小明给出了建议，让小丽把碰头开关开大点，自己开小点，这样就让小丽多用点水，自己少用点也就够了。
2.2.1 生产者
同上面生产者一致
2.2.2 消费者
小明洗澡：
/**
* @author rabbit
* @version 1.0.0
* @Description 工作模式 一个生产者，一个默认的交换机，一个队列，两个消费者
* 需要在consumer消费者端，添加Qos能力以及更改为ACK手动即可让消费者根据自己的能力消费，不是RabbitMQ默认的平均分配了
* @createTime 2022/07/27 19:36:00
*/
public class XMShowerConsumer {
public static final String QUEUE_NAME = "SolarWaterHeater";
//消费者
public static void main(String[] args) throws Exception {
//1、获取连对象、
Connection connection = RabbitCommonConfig.getConnection();
//2、创建channel
Channel channel = connection.createChannel();
channel.basicQos(1);
//3、创建队列-helloworld
/**
* 参数1：queue 指定队列名称
* 参数2：durable 是否开启持久化（true）
* 参数3：exclusive 是否排外（当前队列只能被一个消费者消费）
* 参数4：autoDelete 如果这个队列没有其他消费者在消费，队列自动删除
* 参数5：arguments 指定队列携带的信息
*/
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
//4.开启监听Queue
DefaultConsumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("小明洗澡已用水-第: " + new String(body, "UTF-8"));
//手动ACK（接收信息，指定是否批量操作）
channel.basicAck(envelope.getDeliveryTag(), false);
}
};
/**
* 参数1：queue 指定消费哪个队列
* 参数1：deliverCallback 指定是否ACK（true:收到消息会立即告诉RabbiMQ，false：手动告诉）
* 参数1：cancelCallback 指定消费回调
*/
channel.basicConsume(QUEUE_NAME, false, consumer);
System.out.println("小明开始洗澡-慢慢洗......");
//5、键盘录入，让程序不结束！
System.in.read();
//6、释放资源
channel.close();
connection.close();
}
}小丽洗澡：
/**
* @author rabbit
* @version 1.0.0
* @Description 工作模式 一个生产者，一个默认的交换机，一个队列，两个消费者
* 需要在consumer消费者端，添加Qos能力以及更改为ACK手动即可让消费者根据自己的能力消费，不是RabbitMQ默认的平均分配了
* @createTime 2022/07/27 19:36:00
*/
public class XLShowerConsumer {
public static final String QUEUE_NAME = "SolarWaterHeater";
//消费者
public static void main(String[] args) throws Exception {
//1、获取连对象、
Connection connection = RabbitCommonConfig.getConnection();
//2、创建channel
Channel channel = connection.createChannel();
channel.basicQos(1);
//3、创建队列-helloworld
/**
* 参数1：queue 指定队列名称
* 参数2：durable 是否开启持久化（true）
* 参数3：exclusive 是否排外（当前队列只能被一个消费者消费）
* 参数4：autoDelete 如果这个队列没有其他消费者在消费，队列自动删除
* 参数5：arguments 指定队列携带的信息
*/
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
//4.开启监听Queue
DefaultConsumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
try {
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("小丽洗澡已用水-第: " + new String(body, "UTF-8"));
//手动ACK（接收信息，指定是否批量操作）
channel.basicAck(envelope.getDeliveryTag(), false);
}
};
/**
* 参数1：queue 指定消费哪个队列
* 参数1：deliverCallback 指定是否ACK（true:收到消息会立即告诉RabbiMQ，false：手动告诉）
* 参数1：cancelCallback 指定消费回调
*/
channel.basicConsume(QUEUE_NAME, false, consumer);
System.out.println("小丽开始洗澡-快速洗......");
//5、键盘录入，让程序不结束！
System.in.read();
//6、释放资源
channel.close();
connection.close();
}
}2.2.3 结果验证
生产者：
水龙头放水成功！1升
水龙头放水成功！2升
水龙头放水成功！3升
水龙头放水成功！4升
水龙头放水成功！5升
水龙头放水成功！6升
水龙头放水成功！7升
水龙头放水成功！8升
水龙头放水成功！9升
水龙头放水成功！10升
水龙头放水成功！11升
水龙头放水成功！12升
水龙头放水成功！13升
水龙头放水成功！14升
水龙头放水成功！15升
水龙头放水成功！16升
水龙头放水成功！17升
水龙头放水成功！18升
水龙头放水成功！19升
水龙头放水成功！20升小明：
小明开始洗澡-慢慢洗......
小明洗澡已用水-第: 2升
小明洗澡已用水-第: 7升
小明洗澡已用水-第: 12升
小明洗澡已用水-第: 17升小丽：
小丽开始洗澡-快速洗......
小丽洗澡已用水-第: 1升
小丽洗澡已用水-第: 3升
小丽洗澡已用水-第: 4升
小丽洗澡已用水-第: 5升
小丽洗澡已用水-第: 6升
小丽洗澡已用水-第: 8升
小丽洗澡已用水-第: 9升
小丽洗澡已用水-第: 10升
小丽洗澡已用水-第: 11升
小丽洗澡已用水-第: 13升
小丽洗澡已用水-第: 14升
小丽洗澡已用水-第: 15升
小丽洗澡已用水-第: 16升
小丽洗澡已用水-第: 18升
小丽洗澡已用水-第: 19升
小丽洗澡已用水-第: 20升从结果看：实现了我们期望的结果，小丽用的水多了，小明用的水少，大家都洗好了。
3.模式总结
此模式我们最应该注意的就是平均模式与公平模式的实现，这里是靠消费者的手工确认机制来实现的。
String basicConsume(String queue, boolean autoAck, Consumer callback) throws IOException;
如上代码中的 第二个参数 autoAck，
true的时候，属于自动确认，消费者一旦接收到此消息，就会发回确认消息给Broker，Broker会从队列中删除掉此消息。当有多个消费者注册到同一个queue时，会默认轮询分发。
false的时候，属于手动确认，消费者虽然接收到了消息，还需要执行一个 方法来告诉Broker才行
channel.basicAck(envelope.getDeliveryTag(), false);若没有告诉Broker，Broker还会将此消息再次发送。
同时此种情况下还需要使用此方法：
void basicQos(int prefetchCount) throws IOException;指定每次消费抓取的数量
maximum number of messages that the server will deliver这样我们通过这些改造，实现了，消费者消费的时候，消费一个告诉Broker删除一个，没消费的时候就不要给消费者投递了，最终实现了公平消费。
