SpringBoot集成RabbitMQ-Topic模式_rabbitmq 整合springboot 发送topic消息

一、POM配置文件

 <dependency>
     <groupId>org.springframework.boot</groupId>
     <artifactId>spring-boot-starter-amqp</artifactId>
 </dependency>
1
2
3
4

二、信息发送端

  配置所用的Bean ,接受端和这里一致

@Configuration
public class AmqpConfig {

    /*
    * Bean用于模拟Spring配置文件中的<bean>标签,
    * 用于创建名字为BootDirectExchange的交换机.
    **/
    /*@Bean
    public DirectExchange myChange(){
        return new DirectExchange("BootDirectExchange");
    }*/

    /*
    * 创建一个基于Fanout的交换机
    * 名字为BootFanoutExchange
    * */
    /*@Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("BootFanoutExchange");
    }*/
    @Bean
    public Queue topicQueue(){
        return new Queue("topicQueue");
    }
    @Bean
    public Queue topicQueue2(){
        return new Queue("topicQueue2");
    }
    @Bean
    public TopicExchange topicExchange(){
        return new TopicExchange("BootTopicExchange");
    }
    //创建一个名字为myQueueDirect的队列
    /*@Bean
    public Queue queue(){
        return new Queue("myQueueDirect");
    }*/
    /*@Bean
    public Queue fanoutQueue(){
        return new Queue("fanoutQueue");
    }*/

     /* 将队列绑定到交换机,参数BootRouting为RoutingKey
     * 参数1 为自定义队列对象,参数名queue为自定义队列Bean的id
     * 参数2 为自定义的交换机,参数名myChange为自定义交换机Bean的id
     * */
    /*@Bean("binding")
    public Binding binding(@Qualifier("queue") Queue queue, @Qualifier("myChange")Exchange myChange){
        return BindingBuilder.bind(queue).to(myChange).with("BootRouting").noargs();
    }*/

    //fanout模式绑定
    /*@Bean("binding")
    public Binding binding(@Qualifier("fanoutQueue") Queue queue, @Qualifier("fanoutExchange")FanoutExchange myChange){
        return BindingBuilder.bind(queue).to(myChange);
    }*/

    //topic绑定队列到交换机
    @Bean
    public Binding  topicBinding(Queue topicQueue,TopicExchange topicExchange){
        return BindingBuilder.bind(topicQueue).to(topicExchange).with("Boot.#");
    }
    @Bean
    public Binding  topicBinding2(Queue topicQueue2,TopicExchange topicExchange){
        return BindingBuilder.bind(topicQueue2).to(topicExchange).with("#.text");
    }
}

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

  发送消息类

@Service
public class TopicSend {
    //自动注入Amqp的模板对象
    @Resource
    private AmqpTemplate template;
    public void topicSend(){
        /*
        * 发送消息
        * 参数1: 交换机名称
        * 参数2: Routingkey
        * 参数3: 为具体的消息内容
         * */
        String message = "凭君莫话封侯事,一将功成万骨枯！";
        template.convertAndSend("BootTopicExchange","Boot.text",message);
        System.out.println("发送消息成功:"+message);
    }
}

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

  主方法

@SpringBootApplication
public class ApplicationSend {

	public static void main(String[] args) {
		ApplicationContext ac= SpringApplication.run(ApplicationSend.class, args);
		TopicSend send = (TopicSend)ac.getBean("topicSend");
        send.topicSend();
	}

}
1
2
3
4
5
6
7
8
9
10

三、信息接受端

  接受消息类

@Service
public class TopicReceive {
    //@RabbitListener注解用于标记当前方法为消息监听方法,可以监听某个队列,当队列中有新消息则自动完成接收.
    @RabbitListener(queues ="topicQueue")
    public void receive(String message){
        System.out.println("Boot的topic消息----"+message);
    }

    @RabbitListener(queues ="topicQueue2")
    public void receive2(String message){
        System.out.println("Boot的topic消息2----"+message);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13

  主方法

@SpringBootApplication
public class ApplicationReceive {

	public static void main(String[] args) {
		ApplicationContext ac = SpringApplication.run(ApplicationReceive.class, args);

		TopicReceive receive = (TopicReceive)ac.getBean("topicReceive");
		receive.receive("");
		receive.receive2("");
	}
}
1
2
3
4
5
6
7
8
9
10
11