RabbitMQ_rabbitmq maxconcurrentconsumers

RabbitMQ

rabbitmq创建方式的基本流程：

1. 定义配置信息：主机ip、mq端口、虚拟主机、用户、密码；（普通方式直接定义到类的常量中）（在实际项目中直接配置到yml文件中）

• yml文件中的配置方式

rabbitmq: 
  host: localhopst #服务器地址
  port: 5672
  virtual-host: /name #rabbitmq的虚拟host
  # 当前用户的用户名和密码；
  username: guest
  password: guest
  publisher-confirm-type: correlated # 支持发布订阅
  publisher-return: true # 支持发布返回
  # 虚拟主机名称
  listener: 
    simple: 
      # 设置当前消费者数量（线程数）
      concurrency: 5
      # 设置消费者最大并发数量
      max-concurrency: 10
      # 每个消费者可能未完成的最大未确定消息数量
      prefech: 1
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

2. 创建连接工厂并将配置信息set到连接工厂：

• 普通语法：

	    //创建连接工厂
        ConnectionFactory factory=new ConnectionFactory();
        //设置服务器地址
        factory.setHost(host);
        //设置服务器端口
        factory.setPort(port);
        //设置虚拟主机
        factory.setVirtualHost(virtualHost);
        //设置用户名
        factory.setUsername(userName);
        //设置密码
        factory.setPassword(password);
1
2
3
4
5
6
7
8
9
10
11
12

• 项目语法：

//1.用@Value从yml文件中获取出配置信息到类的变量上
//2.创建方法：连接工厂、rabbitTemplate、单一消费者、多个消费者：
   /**
     * rabbitMQ 连接工厂
     */
    @Bean
    public ConnectionFactory connectionFactory() {
        //创建连接工厂
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        //设置参数
        connectionFactory.setAddresses(hostname + ":" + port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost(virtualHost);
        connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.valueOf(publisherConfirmType.toUpperCase()));
        //返回一个配置好的连接工厂实例
        return connectionFactory;
    }

    /**
     * 配置rabbitTemplate
     */
    @Bean
    //取消springIOC默认单例模式。更改为多例模式
    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    public RabbitTemplate rabbitTemplate() {
        //每次调用都将创建新的rabbitTemplate对象；并将连接工厂传进去
        return new RabbitTemplate(connectionFactory());
    }

    /**
     * 单一消费者
     */
    @Bean(name = "singleListenerContainer")
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory());
        //单一消费者
        factory.setConcurrentConsumers(1);
        //最大消费者数量
        factory.setMaxConcurrentConsumers(1);
        //每个消费者可能未完成的最大未确认消息数量:1.就是处理完一个再去接受下一个信息
        factory.setPrefetchCount(1);
        //确认消息类型:手动
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        return factory;
    }

    /**
     * 多个消费者
     */
    @Bean(name = "multiListenerContainer")
    public SimpleRabbitListenerContainerFactory multiListenerContainerFactory(){
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory());
        //设置当前消费者数量(线程数)
        factory.setConcurrentConsumers(concurrency);
        //设置消费者最大并发数量
        factory.setMaxConcurrentConsumers(maxConcurrency);
        //每个消费者可能未完成的最大未确认消息数量
        factory.setPrefetchCount(prefetch);
        //确认消息类型:手动
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        return factory;
    }
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

3. 项目中可定义枚举或const将路由信息、队列信息、交换机名称信息定义进去；
4. 定义配置类：

//1.定义交换机；
//2.定义队列；
//3.定义交换机绑定队列；
1
2
3

后期完善…