RabbitMq – Work 模式
一、什么是Work模式
如果有几个消息都需要处理，且每个消息的处理时间很长，仅有一个消费者，那么当它在处理一个消息的时候，其他消息就只有等待。
等待有时候是好的，但在程序中并不那么好，当队列中有多个消息待处理，将其分发给多个消费者，当一个消费者在处理的时候，有其他消费者继续消费队列中的消息，便缓解了等待的尴尬。
那么这篇文章将实现一个生产者，多个消费者的模式，实现任务分发：work模式，如图所示。
二、消息确认机制
问题：怎样保证消息不因消费者gg而丢失
处理一个消息可能会花一定的时间，万一还没处理完消费者就gg了…生产者一发送消息，便会将其标记为已删除，故最终的结果是：这条消息没有得到正确的处理。而且，指派给该消费者且尚未处理的所有消息都会gg。
解决策略：取消自动回复机制
为了解决消息的丢失问题，RabbitMQ提供了消息确认机制：message acknowledgments，一个消费者处理完成后，将会回传一个ack给生产者，以表示处理成功，这样生产者才可以将消息删除。
这样即使一个消费者gg了，没有回传ack，那么发送者便会重发消息到队列，如果这时候有其他的消费者服务该队列，那么便会从队列中取出消息并处理。这就保证了消息的不丢失。
自动回复机制：不管是否处理成功，还是失败，都会回复ack。
channel.basicConsume(QUEUE_NAME, true, consumer);
自动恢复机制默认是打开的，在接收端的代码最后：第二个参数为true，表示会自动回复，只要生产发送消息，就会标记删除。所以我们需要将自动回复设置为false。
boolean autoAck = false;
channel.basicConsume(QUEUE_NAME, autoAck, consumer);
这样来保证消息不会因为消费者的gg而丢失了。
那么取消自动回复以后，我们需要手动回复一次：
channel.basicAck(envelope.getDeliveryTag(), false);
注意当前的消息确认机制只适用于同一个channel。
———————
application.yml
#############################################################
############## rabbitmq config ##############################
#############################################################
spring.rabbitmq.host: 127.0.0.1
spring.rabbitmq.port:
spring.rabbitmq.username: admin
spring.rabbitmq.password: admin
package com.maozw.mq.config; import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope; /**
* @author MAOZW
* @Description: Broker:它提供一种传输服务,它的角色就是维护一条从生产者到消费者的路线，保证数据能按照指定的方式进行传输,
* Exchange：消息交换机,它指定消息按什么规则,路由到哪个队列。
* Queue:消息的载体,每个消息都会被投到一个或多个队列。
* Binding:绑定，它的作用就是把exchange和queue按照路由规则绑定起来.
* Routing Key:路由关键字,exchange根据这个关键字进行消息投递。
* vhost:虚拟主机,一个broker里可以有多个vhost，用作不同用户的权限分离。
* Producer:消息生产者,就是投递消息的程序.
* Consumer:消息消费者,就是接受消息的程序.
* Channel:消息通道,在客户端的每个连接里,可建立多个channel.
* @date 2018/11/26 14:33
*/
@Configuration
public class RabbitConfig { private static final Logger LOGGER = LoggerFactory.getLogger(RabbitConfig.class); @Value("${spring.rabbitmq.host}")
private String rabbitmqHost; @Value("${spring.rabbitmq.port}")
private int rabbitmqPort; @Value("${spring.rabbitmq.username}")
private String userName; @Value("${spring.rabbitmq.password}")
private String password; public static final String EXCHANGE_A = "my-mq-exchange_A";
public static final String EXCHANGE_B = "my-mq-exchange_B";
public static final String EXCHANGE_C = "my-mq-exchange_C"; public static final String QUEUE_A = "QUEUE_A";
public static final String QUEUE_WORK = "QUEUE_WORK";
public static final String QUEUE_C = "QUEUE_C"; public static final String ROUTINGKEY_A = "spring-boot-routingKey_A";
public static final String ROUTINGKEY_B = "spring-boot-routingKey_B";
public static final String ROUTINGKEY_C = "spring-boot-routingKey_C"; @Bean
public ConnectionFactory connectionFactory() {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
connectionFactory.setUsername(userName);
connectionFactory.setPassword(password);
connectionFactory.setVirtualHost("/vir_simple");
connectionFactory.setPublisherConfirms(false);
return connectionFactory;
} public static ConnectionFactory getConnectionFactory() {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
connectionFactory.setUsername("admin");
connectionFactory.setPassword("admin");
connectionFactory.setVirtualHost("/vir_simple");
connectionFactory.setPublisherConfirms(false);
return connectionFactory;
}
}
生产者
package com.maozw.mq.work; import com.maozw.mq.config.RabbitConfig;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController; import java.io.IOException; /**
* work 模式
* 两种分发： 轮询分发 + 公平分发
* 轮询分发：消费端：自动确认消息；boolean autoAck = true;
* 公平分发: 消费端：手动确认消息 boolean autoAck = false; channel.basicAck(envelope.getDeliveryTag(),false);
* @author MAOZW
* @Description: ${todo}
* @date 2018/11/26 15:06
*/
@RestController
@RequestMapping("/work")
public class WorkProducer {
private static final Logger LOGGER = LoggerFactory.getLogger(WorkProducer.class);
@Autowired
RabbitConfig rabbitConfig; @RequestMapping("/send")
public String send() throws IOException {
ConnectionFactory connectionFactory = rabbitConfig.connectionFactory();
Connection connection = connectionFactory.createConnection();
Channel channel = connection.createChannel(false);
//创建队列申明
channel.queueDeclare(RabbitConfig.QUEUE_WORK, false, false, false, null); /**
* 每个消费者 发送确认消息之前，消息队列不会发送下一个消息给消费者，一次只处理一个消息
* 自动模式无需设置下面设置
*/
int prefetchCount = ;
channel.basicQos(prefetchCount); String Hello = ">>>> Hello Simple <<<<";
for (int i = ; i < ; i++) {
String message = Hello + i;
channel.basicPublish("", RabbitConfig.QUEUE_WORK, null, message.getBytes());
LOGGER.info("生产消息： " + message);
}
return "OK";
}
}
消费者
package com.maozw.mq.work; import com.maozw.mq.config.RabbitConfig;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory; import java.io.IOException; /**
* @author MAOZW
* @Description: ${todo}
* @date 2018/11/26 15:06
*/ public class WorkConsumer2 {
private static final Logger LOGGER = LoggerFactory.getLogger(WorkConsumer2.class); public static void main(String[] args) throws IOException {
ConnectionFactory connectionFactory = RabbitConfig.getConnectionFactory();
Connection connection = connectionFactory.createConnection();
Channel channel = connection.createChannel(false);
//创建队列申明
channel.queueDeclare(RabbitConfig.QUEUE_WORK, false, false, false, null); /**
* 改变分发规则
*/
channel.basicQos(); DefaultConsumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
super.handleDelivery(consumerTag, envelope, properties, body);
System.out.println("[2] 接口数据 ： " + new String(body, "utf-8")); try {
Thread.sleep();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
System.out.println("[2] done!");
//消息应答：手动回执，手动确认消息
channel.basicAck(envelope.getDeliveryTag(),false);
}
}
};
//监听队列
/**
* autoAck 消息应答
* 默认轮询分发打开：true ：这种模式一旦rabbitmq将消息发送给消费者，就会从内存中删除该消息，不关心客户端是否消费正常。
* 使用公平分发需要关闭autoAck:false 需要手动发送回执
*/
boolean autoAck = false;
channel.basicConsume(RabbitConfig.QUEUE_WORK,autoAck, consumer);
} }
