RabbitMQ 开发随笔

SpringBoot 中配置 RabbitMQ 使用自定义消息转换器

业务之间大多数数据都是以 JSON 的数据格式进行传输的,即生产者服务将 JSON 类型的数据发送到对应的队列, 而消费端从队列中接收到的数据类型也是 JSON 类型,为了方便将 Java 对象转为 JSON 类型的数据来传输,此时可以使用 Spring 内置的 Jackson2JsonMessageConverter 消息转换器,具体代码如下:

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
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMqConfig {

/**
* RabbitMQ的管理对象
*
* @param connectionFactory
* @return
*/
@Bean
public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
return rabbitAdmin;
}

/**
* RabbitMq的消息转换器
*
* @return
*/
@Bean
public MessageConverter jsonMessageConverter() {
Jackson2JsonMessageConverter messageConverter = new Jackson2JsonMessageConverter();
return messageConverter;
}

/**
* RabbitMq的模版
*
* @return
*/
@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory, MessageConverter messageConverter) {
RabbitTemplate template = new RabbitTemplate(connectionFactory);
// 设置发送消息时所用的消息转换器
template.setMessageConverter(messageConverter);
return template;
}

/**
* RabbitMq的监听容器工厂
*
* @return
*/
@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory, MessageConverter messageConverter) {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setConnectionFactory(connectionFactory);
// 设置线程数
factory.setConcurrentConsumers(3);
// 最大线程数
factory.setMaxConcurrentConsumers(10);
// 设置接收消息时所用的消息转换器
factory.setMessageConverter(messageConverter);
return factory;
}

}

Java 消息队列任务的平滑关闭分析