Rabbitmq - rabbitmq Listener监听

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

listener:
simple:
prefetch: 1 # 一次拉取的数量
concurrency: 5 # 消费端的监听个数(即@RabbitListener开启几个线程去处理数据。)
max-concurrency: 10 # 消费端的监听最大个数
acknowledge-mode: manual
retry:
multiplier: 1
max-attempts: 3
enabled: true
direct:
retry:
enabled: true
max-attempts: 3
acknowledge-mode: manual
auto-startup: true
type: simple

1
2
3
listener.type 表示监听的类型 主要有两种 simple和direct 对于的监听容器是
SimpleMessageListenerContainer和DirectMessageListenerContainer

1
2
3
4
5
listener.simple.prefetch:
每个customer会在MQ预取一些消息放入内存的LinkedBlockingQueue中进行消费，这个值越高，消息传递的越快，
但非顺序处理消息的风险更高。如果ack模式为none，则忽略。如有必要，将增加此值以匹配txSize或messagePerAck。
不过在有些情况下，尤其是处理速度比较慢的大消息，消息可能在内存中大量堆积，消耗大量内存；以及对于一些严格要求顺序的消息，prefetch的值应当设置为1。

1
2
3
4
5
6
7
8
acknowledge-mode：表示消费端收到消息后的确认方式。
有三种确认方式：
自动确认：acknowledge="none"
手动确认：acknowledge="manual"
根据异常情况确认：acknowledge="auto"
其中自动确认是指，当消息一旦被Consumer接收到，则自动确认收到，并将相应 message 从 RabbitMQ 的消息缓存中移除。但是在实际业务处理中，很可能消息接收到，业务处理出现异常，那么该消息就会丢失。
如果设置了手动确认方式，则需要在业务处理成功后，调用channel.basicAck()，手动签收，如果出现异常，则调用channel.basicNack()方法，让其自动重新发送消息。

1
2
3
4
5
concurrency：
concurrency =1，即每个Listener容器将开启一个线程去处理消息。
可以在 @RabbitListener（concurrency = "3"）直接配置当前监听器启动线程，如果在Listener配置了exclusive参数，即确定此容器中的单个customer是否具有对队列的独占访问权限。如果为true，则容器的并发性必须为1。
如果配置了exclusive=true,但是concurrency>1则会抛错

1
2
3
4
prefetch和concurrency
若一个消费者配置prefetch=10，concurrency=2，即会开启2个线程去消费消息，每个线程都会抓取10个线程到内存中（注意不是两个线程去共享内存中抓取的消息）
max-concurrency:： 表示最大能启动的线程数

1
2
3
4
5
6
7
8
9
retry:
表示消息被消费者拒收后重试发送或者因为异常原因消息重试发送
multiplier：重试间隔乘法策略
max-attempts： 是最大重试次数，默认是三
enabled： 是否开启重试
initial-interval： 初始化重试次数间隔
max-interval 最大重试间隔
stateless：重试是无状态的还是有状态的。

1
2
3
autoStartup:
当autoStartup为false的时候，监听容器就不会自动启动，然后我们可以通过使用单个容器的ID，调用RabbitListenerEndpointRegistry类的getListenerContainer（String id）方法来获得对单个容器的引用，并执行strat方法，启动容器。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

@RabbitListener(bindings = {
@QueueBinding(
value = @Queue(value = "${rabbitmq.queue.routing.beijing}", durable = "true", autoDelete = "false"),
exchange = @Exchange(
value = "${rabbitmq.exchange.routing}",
durable = "true",
type = ExchangeTypes.TOPIC),
key = "china.#")}, concurrency = "3", exclusive = false,id = "autoStart",autoStartup = "false")
public void receive(@Payload String msg, Message message,Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException {
System.out.println(String.format("mesage:%s", message.getBody()));
System.out.println("路由监听接受到发送者发送的信息：" + msg);
// 确认消息
//
channel.basicAck(deliveryTag, false);
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
public void stopContainer(String containerId){
//得到容器的对象
MessageListenerContainer container = registry.getListenerContainer(containerId);
//判断容器状态
if(container.isRunning()){
//开启容器
container.stop();
System.out.println("关闭容器");
}
}
public void startContainer(String containerId){
//得到容器的对象
MessageListenerContainer container = registry.getListenerContainer(containerId);
//判断容器状态
if(!container.isRunning()){
//开启容器
container.start();
System.out.println("开启容器");
}
}

1
2
3
4
5
6
7
8
9
public class Message implements Serializable {
private final MessageProperties messageProperties;
private final byte[] body;
public Message(byte[] body, MessageProperties messageProperties) { //NOSONAR
this.body = body; //NOSONAR
this.messageProperties = messageProperties;
}
}

1
2
3
MessageProperties // 消息属性
byte[] body // 消息内容

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
public class SimpleMessageConverter extends WhiteListDeserializingMessageConverter implements BeanClassLoaderAware {
@Override
public Object fromMessage(Message message) throws MessageConversionException {
Object content = null;
MessageProperties properties = message.getMessageProperties();
if (properties != null) {
String contentType = properties.getContentType();
if (contentType != null && contentType.startsWith("text")) {
String encoding = properties.getContentEncoding();
if (encoding == null) {
encoding = this.defaultCharset;
}
try {
content = new String(message.getBody(), encoding);
}
catch (UnsupportedEncodingException e) {
throw new MessageConversionException(
"failed to convert text-based Message content", e);
}
}
else if (contentType != null &&
contentType.equals(MessageProperties.CONTENT_TYPE_SERIALIZED_OBJECT)) {
try {
content = SerializationUtils.deserialize(
createObjectInputStream(new ByteArrayInputStream(message.getBody()), this.codebaseUrl));
}
catch (IOException | IllegalArgumentException | IllegalStateException e) {
throw new MessageConversionException(
"failed to convert serialized Message content", e);
}
}
}
if (content == null) {
content = message.getBody();
}
return content;
}
/**
* Creates an AMQP Message from the provided Object.
*/
@Override
protected Message createMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
byte[] bytes = null;
if (object instanceof byte[]) {
bytes = (byte[]) object;
messageProperties.setContentType(MessageProperties.CONTENT_TYPE_BYTES);
}
else if (object instanceof String) {
try {
bytes = ((String) object).getBytes(this.defaultCharset);
}
catch (UnsupportedEncodingException e) {
throw new MessageConversionException(
"failed to convert to Message content", e);
}
messageProperties.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN);
messageProperties.setContentEncoding(this.defaultCharset);
}
else if (object instanceof Serializable) {
try {
bytes = SerializationUtils.serialize(object);
}
catch (IllegalArgumentException e) {
throw new MessageConversionException(
"failed to convert to serialized Message content", e);
}
messageProperties.setContentType(MessageProperties.CONTENT_TYPE_SERIALIZED_OBJECT);
}
if (bytes != null) {
messageProperties.setContentLength(bytes.length);
return new Message(bytes, messageProperties);
}
throw new IllegalArgumentException(getClass().getSimpleName()
+ " only supports String, byte[] and Serializable payloads, received: " + object.getClass().getName());
}
}