深入剖析 RabbitMQ —— Spring 框架下实现 AMQP 高级消息队列协议（上）

1 **********************pom *************************
 2 <project>
 3 .............
 4 <dependency>
 5 <groupId>org.springframework.boot</groupId>
 6 <artifactId>spring-boot-starter-amqp</artifactId>
 7 <version>2.0.5.RELEASE</version>
 8 </dependency>
 9 </project>
10 
11 **************** application.yml ****************
12 spring:
13 application:
14 name: rabbitMqProducer
15 rabbitmq:
16 host: localhost 
17 port: 5672
18 username: admin
19 password: 12345678
20 virtual-host: /LeslieHost

 1 @Configuration
 2 public class ConnectionConfig {
 3 @Value("${spring.rabbitmq.host}")
 4 public String host;
 5 
 6 @Value("${spring.rabbitmq.port}")
 7 public int port;
 8 
 9 @Value("${spring.rabbitmq.username}")
10 public String username;
11 
12 @Value("${spring.rabbitmq.password}")
13 public String password;
14 
15 @Value("${spring.rabbitmq.virtual-host}")
16 public String virtualHost;
17 
18 @Bean
19 public ConnectionFactory getConnectionFactory(){
20 CachingConnectionFactory factory=new CachingConnectionFactory();
21 factory.setHost(host);
22 factory.setPort(port);
23 factory.setUsername(username);
24 factory.setPassword(password);
25 factory.setVirtualHost(virtualHost);
26 return factory;
27 }
28 }
29 
30 @Configuration
31 public class BindingConfig {
32 public final static String first="direct.first";
33 public final static String second="direct.second";
34 public final static String Exchange_NAME="directExchange";
35 public final static String RoutingKey1="directKey1";
36 public final static String RoutingKey2="directKey2";
37 
38 @Bean
39 public Queue queueFirst(){
40 return new Queue(first);
41 }
42 
43 @Bean
44 public Queue queueSecond(){
45 return new Queue(second);
46 }
47 
48 @Bean
49 public DirectExchange directExchange(){
50 return new DirectExchange(Exchange_NAME,true,true);
51 }
52 
53 //利用BindingBuilder绑定Direct与queueFirst
54 @Bean
55 public Binding bindingExchangeFirst(Queue queueFirst, DirectExchange directExchange){
56 return BindingBuilder.bind(queueFirst).to(directExchange).with(RoutingKey1);
57 }
58 
59 //利用BindingBuilder绑定Direct与queueSecond
60 @Bean
61 public Binding bindingExchangeSecond(Queue queueSecond, DirectExchange directExchange){ 
62 return BindingBuilder.bind(queueSecond).to(directExchange).with(RoutingKey2);
63 } 
64 }
65 
66 @Controller
67 @RequestMapping("/producer")
68 public class ProducerController {
69 @Autowired
70 private RabbitTemplate template;
71 
72 @RequestMapping("/send")
73 public void send() {
74 for(int n=0;n<100;n++){ 
75 
76 template.convertAndSend(BindingConfig.Exchange_NAME,BindingConfig.RoutingKey1,"I'm the first queue! "+String.valueOf(n),getCorrelationData());
77 template.convertAndSend(BindingConfig.Exchange_NAME,BindingConfig.RoutingKey2,"I'm the second queue! "+String.valueOf(n),getCorrelationData());
78 }
79 }
80 
81 private CorrelationData getCorrelationData(){
82 return new CorrelationData(UUID.randomUUID().toString());
83 }
84 }

 1 @Configuration
 2 public class ConnectionConfig {
 3 @Value("${spring.rabbitmq.host}")
 4 public String host;
 5 
 6 @Value("${spring.rabbitmq.port}")
 7 public int port;
 8 
 9 @Value("${spring.rabbitmq.username}")
10 public String username;
11 
12 @Value("${spring.rabbitmq.password}")
13 public String password;
14 
15 @Value("${spring.rabbitmq.virtual-host}")
16 public String virtualHost;
17 
18 @Bean
19 public ConnectionFactory getConnectionFactory(){
20 CachingConnectionFactory factory=new CachingConnectionFactory();
21 factory.setHost(host);
22 factory.setPort(port);
23 factory.setUsername(username);
24 factory.setPassword(password);
25 factory.setVirtualHost(virtualHost);
26 return factory;
27 }
28 }
29 
30 @Configuration
31 public class BindingConfig {
32 public final static String first="direct.first";
33 public final static String Exchange_NAME="directExchange";
34 public final static String RoutingKey1="directKey1";
35 
36 @Bean
37 public Queue queueFirst(){
38 return new Queue(first);
39 }
40 
41 @Bean
42 public DirectExchange directExchange(){
43 return new DirectExchange(Exchange_NAME);
44 }
45 
46 //利用BindingBuilder绑定Direct与queueFirst
47 @Bean
48 public Binding bindingExchangeFirst(Queue queueFirst, DirectExchange directExchange){
49 return BindingBuilder.bind(queueFirst).to(directExchange).with(RoutingKey1);
50 } 
51 }
52 
53 @Configuration
54 @RabbitListener(queues="direct.first")
55 public class RabbitMqListener {
56 
57 @RabbitHandler
58 public void handler(String message){
59 System.out.println(message);
60 }
61 }
62 
63 @SpringBootApplication
64 public class App {
65 
66 public static void main(String[] args){
67 SpringApplication.run(App.class, args);
68 }
69 }

 1 @Configuration
 2 public class ConnectionConfig {
 3 @Value("${spring.rabbitmq.host}")
 4 public String host;
 5 
 6 @Value("${spring.rabbitmq.port}")
 7 public int port;
 8 
 9 @Value("${spring.rabbitmq.username}")
10 public String username;
11 
12 @Value("${spring.rabbitmq.password}")
13 public String password;
14 
15 @Value("${spring.rabbitmq.virtual-host}")
16 public String virtualHost;
17 
18 @Bean
19 public ConnectionFactory getConnectionFactory(){
20 CachingConnectionFactory factory=new CachingConnectionFactory();
21 factory.setHost(host);
22 factory.setPort(port);
23 factory.setUsername(username);
24 factory.setPassword(password);
25 factory.setVirtualHost(virtualHost);
26 return factory;
27 }
28 }
29 
30 @Configuration
31 @RabbitListener(bindings=@QueueBinding(
32 exchange=@Exchange(value="directExchange"),
33 value=@Queue(value="direct.second"),
34 key="directKey2"))
35 public class RabbitMqListener {
36 
37 @RabbitHandler
38 public void handler(String message){
39 System.out.println(message);
40 }
41 }
42 
43 @SpringBootApplication
44 public class App {
45 
46 public static void main(String[] args){
47 SpringApplication.run(App.class, args);
48 }
49 }