SpringBoot集成RabbitMQ_activimq 指定一次消费数量

项目中用到rabbitMQ来削流量峰值，简单记录一下springboot如何集成rabbitMQ的。

SpringBoot集成RabbitMQ

• 第一步：先下载erlang和rabbitMQ-server；
• 第二步：SpringBoot的pom中添加依赖；

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

• 第三步：在application.properties中进行配置；

#rabbitmq
spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=/
#表示消费者数量，如果为1相当于是串行
spring.rabbitmq.listener.simple.concurrency=10
spring.rabbitmq.listener.simple.max-concurrency=10
#表示每次从队列中取，取多少个
spring.rabbitmq.listener.simple.prefetch=1
#消费者自动启动
spring.rabbitmq.listener.simple.auto-startup=true
#消费者消费失败后把数据重新加入到队列中去
spring.rabbitmq.listener.simple.default-requeue-rejected=true
#下面是重试的配置，分别表示：
#允许重试，初始1秒重试一次，最多重试3次，最大间隔是10秒
spring.rabbitmq.template.retry.enabled=true 
spring.rabbitmq.template.retry.initial-interval=1000
spring.rabbitmq.template.retry.max-attempts=3
spring.rabbitmq.template.retry.max-interval=10000
#multiplier表示如果前一次重试间隔为i秒，下一次为i * multiplier秒
spring.rabbitmq.template.retry.multiplier=1.0
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

• 第四步：写rabitmq的config文件；

交换机Exchange一共有四种模式，分别为：Direct模式，Topic模式，Fanout模式，Header模式。这里统一给出配置，之后再分别演示其作用。

@Configuration
public class MQConfig {

    public static final String MIAOSHA_QUEUE = "miaosha.queue";
    public static final String QUEUE = "queue";
    public static final String TOPIC_QUEUE1 = "topic.queue1";
    public static final String TOPIC_QUEUE2 = "topic.queue2";
    public static final String HEADER_QUEUE = "header.queue";
    public static final String TOPIC_EXCHANGE = "topicExchage";
    public static final String FANOUT_EXCHANGE = "fanoutxchage";
    public static final String HEADERS_EXCHANGE = "headersExchage";

    /**
     * Direct模式 交换机Exchange
     */
    @Bean
    public Queue queue() {
        return new Queue(QUEUE, true);
    }//队列参数的意义是：名称+是否要初始化

    /**
     * Topic模式 交换机Exchange
     */
    @Bean
    public Queue topicQueue1() {
        return new Queue(TOPIC_QUEUE1, true);
    }

    @Bean
    public Queue topicQueue2() {
        return new Queue(TOPIC_QUEUE2, true);
    }

    @Bean
    public TopicExchange topicExchage() {
        return new TopicExchange(TOPIC_EXCHANGE);
    }

    @Bean
    public Binding topicBinding1() {
        return BindingBuilder.bind(topicQueue1()).to(topicExchage()).with("topic.key1");
    }

    @Bean
    public Binding topicBinding2() {
        return BindingBuilder.bind(topicQueue2()).to(topicExchage()).with("topic.#");
    }

    /**
     * Fanout模式 交换机Exchange
     */
    @Bean
    public FanoutExchange fanoutExchage() {
        return new FanoutExchange(FANOUT_EXCHANGE);
    }

    @Bean
    public Binding FanoutBinding1() {
        return BindingBuilder.bind(topicQueue1()).to(fanoutExchage());
    }

    @Bean
    public Binding FanoutBinding2() {
        return BindingBuilder.bind(topicQueue2()).to(fanoutExchage());
    }

    /**
     * Header模式 交换机Exchange
     */
    @Bean
    public HeadersExchange headersExchage() {
        return new HeadersExchange(HEADERS_EXCHANGE);
    }

    @Bean
    public Queue headerQueue1() {
        return new Queue(HEADER_QUEUE, true);
    }

    @Bean
    public Binding headerBinding() {
        Map<String, Object> map = new HashMap<String, Object>();
        map.put("header1", "value1");
        map.put("header2", "value2");
        return BindingBuilder.bind(headerQueue1()).to(headersExchage()).whereAll(map).match();
    }
}
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
78
79
80
81
82
83
84
85
86
87

• 第五步：写生产者sender和消费者receiver；(以Direct模式为例)

生产者：

@Service
public class MQSender {

    private static Logger log = LoggerFactory.getLogger(MQSender.class);

    @Autowired
    AmqpTemplate amqpTemplate; //操作queue的工具类

    public void send(Object message) {
        String msg = RedisService.beanToString(message);
        log.info("send message:" + msg);
        amqpTemplate.convertAndSend(MQConfig.QUEUE, msg);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

消费者：

@Service
public class MQReceiver {

    private static Logger log = LoggerFactory.getLogger(MQReceiver.class);

    @RabbitListener(queues = MQConfig.QUEUE)
    public void receive(String message) {
        log.info("receive message:" + message);
    }
}
1
2
3
4
5
6
7
8
9
10

• 第6步：Controller层验证实现的消息队列

@Controller
@RequestMapping("/demo")
public class DemoController {
    
    @Autowired
    MQSender sender;

    @RequestMapping("/mq")
    @ResponseBody
    public Result<String> mq() {
        sender.send("hello,imooc");
        return Result.success("Hello，world");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

结果展示
在浏览器中输入url，返回json串：

在控制台上可以看到log打印出来了：

rabbitmq配置、运行成功。

4种模式讲解

• 第一种是Direct模式，在上面已举过例子了。调用MQSender中的sender（）方法，会直接自动加入到MQReceiver的队列Queue中，然后从队列中取出运行方法。
• 第二种是Topic模式。该模式下把信息发送给Exchange交换机，交换机会根据路由来选择相关队列，并执行队列中的方法。前提是要先把队列和交换机绑定并指定好路由，在上面的config中代码中有。
  举例：这里，我们用两个队列来演示。

生产者

@Service
public class MQSender {

    private static Logger log = LoggerFactory.getLogger(MQSender.class);

    @Autowired
    AmqpTemplate amqpTemplate; //操作queue的工具类

    public void sendTopic(Object message) {
        String msg = RedisService.beanToString(message);
        log.info("send topic message:" + msg);
        amqpTemplate.convertAndSend(MQConfig.TOPIC_EXCHANGE, "topic.key1", msg + "1");
        amqpTemplate.convertAndSend(MQConfig.TOPIC_EXCHANGE, "topic.key2", msg + "2");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

消费者

@Service
public class MQReceiver {

    private static Logger log = LoggerFactory.getLogger(MQReceiver.class);

    @RabbitListener(queues = MQConfig.TOPIC_QUEUE1)
    public void receiveTopic1(String message) {
        log.info(" topic  queue1 message:" + message);
    }

    @RabbitListener(queues = MQConfig.TOPIC_QUEUE2)
    public void receiveTopic2(String message) {
        log.info(" topic  queue2 message:" + message);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

控制层

@Controller
@RequestMapping("/demo")
public class DemoController {

    @Autowired
    MQSender sender;

    @RequestMapping("/mq/topic")
    @ResponseBody
    public Result<String> topic() {
        sender.sendTopic("hello,imooc");
        return Result.success("Hello，world");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

结果展示

• 第三种是Fanout模式。这个模式类似于广播模式，和FanoutExchange绑定的队列都能收到消息。
  举例：这里也是用两个队列演示。绑定的过程在上面的config里。

生产者

@Service
public class MQSender {

    private static Logger log = LoggerFactory.getLogger(MQSender.class);

    @Autowired
    AmqpTemplate amqpTemplate; //操作queue的工具类

    public void sendFanout(Object message) {
        String msg = RedisService.beanToString(message);
        log.info("send fanout message:" + msg);
        amqpTemplate.convertAndSend(MQConfig.FANOUT_EXCHANGE, "", msg);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

消费者

@Service
public class MQReceiver {

    private static Logger log = LoggerFactory.getLogger(MQReceiver.class);

    @RabbitListener(queues = MQConfig.TOPIC_QUEUE1)
    public void receiveTopic1(String message) {
        log.info(" topic  queue1 message:" + message);
    }

    @RabbitListener(queues = MQConfig.TOPIC_QUEUE2)
    public void receiveTopic2(String message) {
        log.info(" topic  queue2 message:" + message);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

控制层

@Controller
@RequestMapping("/demo")
public class DemoController {

    @Autowired
    MQSender sender;

    @RequestMapping("/mq/fanout")
    @ResponseBody
    public Result<String> fanout() {
        sender.sendFanout("hello,imooc");
        return Result.success("Hello，world");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

结果展示

• 第四种是Header模式。Header模式和Topic模式有些相似。Topic模式是通过路由来找到相应的队列，而Header模式是通过匹配键值对来找到相关队列。匹配键值对时也可以选择匹配某一个键值对或者匹配所有键值对才能找到相关队列。
  举例：

生产者
由于我在config中设置的是全部匹配才能进入队列，这里举了三个发送方法的例子，验证匹配时必须要全部匹配。

@Service
public class MQSender {

    private static Logger log = LoggerFactory.getLogger(MQSender.class);

    @Autowired
    AmqpTemplate amqpTemplate; //操作queue的工具类

    public void sendHeader(Object message) {
        String msg = RedisService.beanToString(message);
        log.info("send header message:" + msg);
        MessageProperties properties = new MessageProperties();
        properties.setHeader("header1", "value1");
        properties.setHeader("header2", "value2");
        Message obj = new Message(msg.getBytes(), properties);
        amqpTemplate.convertAndSend(MQConfig.HEADERS_EXCHANGE, "", obj);
    }

    public void sendHeader2(Object message) {
        String msg = RedisService.beanToString(message);
        log.info("send header message:" + msg);
        MessageProperties properties = new MessageProperties();
        properties.setHeader("header1", "value1");
        Message obj = new Message(msg.getBytes(), properties);
        amqpTemplate.convertAndSend(MQConfig.HEADERS_EXCHANGE, "", obj);
    }

    public void sendHeader3(Object message) {
        String msg = RedisService.beanToString(message);
        log.info("send header message:" + msg);
        MessageProperties properties = new MessageProperties();
        properties.setHeader("header1", "value2");
        properties.setHeader("header2", "value3");
        Message obj = new Message(msg.getBytes(), properties);
        amqpTemplate.convertAndSend(MQConfig.HEADERS_EXCHANGE, "", obj);
    }
}
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

消费者

@Service
public class MQReceiver {

    private static Logger log = LoggerFactory.getLogger(MQReceiver.class);

    @RabbitListener(queues = MQConfig.HEADER_QUEUE)
    public void receiveHeaderQueue(byte[] message) {
        log.info(" header  queue message:" + new String(message));
    }
}
1
2
3
4
5
6
7
8
9
10

控制层

@Controller
@RequestMapping("/demo")
public class DemoController {

    @Autowired
    MQSender sender;

    @RequestMapping("/mq/header")
    @ResponseBody
    public Result<String> header() {
        sender.sendHeader("hello,imooc");
        sender.sendHeader2("hello,imooc2");
        sender.sendHeader3("hello,imooc3");
        return Result.success("Hello，world");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

结果展示

可以看到，我发送了三个消息，只有一个消息进入消息队列并完成了，因为只有此消息的属性键值对和设置的Header属性键值对是一致的。

至此，四种模式如何使用记录完毕。