rabbitmq安装、配置和收发消息实例代码、性能和高可用

╰半橙微兮° 2022-02-24 17:58 101阅读 0赞

## rabbitMq的安装（mac上）与使用 ##

brew install rabbitmq  
cd  /usr/local/Cellar/rabbitmq/3.7.12  
cd sbin  
./rabbitmq-server  
浏览器输入localhost:15672,账号密码全输入guest即可登录。  
https://blog.csdn.net/u010046908/article/details/54773323

增加rabbitMq的用户springcloud  
![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2gyNjA0Mzk2NzM5_size_16_color_FFFFFF_t_70][]  
为springcloud授权vhost、读写等权限，如下图

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2gyNjA0Mzk2NzM5_size_16_color_FFFFFF_t_70 1][]

tags为administrator 表示最大权限

创建queue，如下图

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2gyNjA0Mzk2NzM5_size_16_color_FFFFFF_t_70 2][]

还需要新建exchange，名称Exchange-Demo，并绑定对列demo，RoutingKey为RK。

## mq的配置实例 ##

建springboot项目方法可以借鉴：https://blog.csdn.net/forezp/article/details/70341651

producer项目  
pom增加maven依赖

<dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-amqp</artifactId>
            </dependency>

application.properties配置

server.port=8080
    spring.application.name=mq-producer
    #注意 这里只能使用localhost或127.0.0.1；如果需要使用虚拟ip进行远程访问，需要
    #修改abbitmq.config配置文件，增加[{rabbit,[{loopback_users,[]}]}]-未测试
    spring.rabbitmq.host=localhost
    #注意这里不是15672，和页面用的不是同一个端口
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=springcloud
    spring.rabbitmq.password=springcloud

producer服务

@Component("rabbitMqProducer")
    public class RabbitMqProducer {
        @Autowired
        private AmqpTemplate template;
    
        public void publish(String msg) {
            //这里 Exchange-Demo是上面新建的exchange
            template.convertAndSend("Exchange-Demo", "RK", msg);
            template.convertAndSend( msg);
        }
    }

controller服务：

@RestController
    public class HelloController {
    
        @Autowired
        RabbitMqProducer rabbitMqProducer;
    
        @RequestMapping("/")
        public String index() {
            rabbitMqProducer.publish("imessage");
            return "Greetings from Spring Boot!";
        }
    }

访问http://localhost:8080/，即可向队列中存入消息

新建一个springboot项目作为consumer  
pom增加maven依赖

<dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-amqp</artifactId>
            </dependency>

application.properties内容：

server.port=8081
    spring.application.name=mq
    spring.rabbitmq.host=localhost
    #注意这里不是15672，和页面用的156782不是同一个端口，是默认的5672
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=springcloud
    spring.rabbitmq.password=springcloud

增加consumer的代码

@Component("rabbitMqConsumer")
    public class RabbitMqConsumer {
        //监听队列名称为demo
        @RabbitListener(queues = "demo")
        public void consumeDemoDirectQueue0(String msg, Channel channel, Message message) throws IOException {
            //项目启动后，会自动消耗名称为demo的queue中的消息
            System.out.println("RabbitMqConsumer.consumeDemoDirectQueue0(): msg=" + msg);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }
    }

注意：@RabbitListener(queues = "demo")中的demo是实际queue的名称，demo对列中有消息的时候，RabbitMqConsumer会自动消费

## mq详细使用问题： ##

1怎么将消息发送到特定的多个队列，单个消息怎么让多个comsumer消费  
建立类型位fanout模式的exchange，假设名称为e.f1,让e.f1绑定多个queue即可。  
e.f1绑定queue的时候让填routingkey,其实在fanout模式routingkey没有任何作用。

2怎么确保消息接收方接收到了消息，即消息不丢失  
开启手动确认机制，即配置文件中增加如下内容：  
\# 开启手动ACK  
spring.rabbitmq.listener.simple.acknowledge-mode=manual  
此时只有消费者手动调用channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);，mq才会删除该消息；  
但是可能出现虽然consumer已经ack了，但是mq没有收到，这是如果consumer发生了重启，消息会再次发送到consumer，即consumer需要做对消息的去重处理

##    mq配置解析 ##

application.properties：  
spring.application.name=rabbit-mq-consumer  
spring.rabbitmq.host=192.168.2.123  
spring.rabbitmq.port=5671  
spring.rabbitmq.username=admin  
spring.rabbitmq.password=123456

\# channel 为逻辑上的连接，多个 channel 共用一个物理TCP连接，默认为 channel  
spring.rabbitmq.cache.connection.mode=channel  
\# 开启手动ACK，未手动ack的消息不会被删除  
spring.rabbitmq.listener.simple.acknowledge-mode=manual  
\# consumer 所能保持的未 ACK 的消息数量，默认为 250，超过该数量，不再向consumer发送消息  
spring.rabbitmq.listener.simple.prefetch=250  
\# 开启发送消息到exchange确认机制  
spring.rabbitmq.publisher-confirms=true  
\# 开启发送消息到 exchange 但是 exchange 没有和队列绑定的确认机制  
spring.rabbitmq.publisher-returns=true

## 注意事项 ##

1、配置 spring.rabbitmq.cache.connection.mode 为 connection 后程序无法创建 Exchange、Queue 和  
 Binding  
2、配置 spring.rabbitmq.listener.simple.acknowledge-mode 为 manual 后，需要手动调用 channel.basicAck，  
否则会导致消息处于 Unacked 状态，并且 Consumer 连接断开或者重启后所有处于 Unacked 状态的消息会重新  
进入队列。  
3、强烈建议在 Consumer 和 Producer 中同时声明 Queue、Exchange 和 Binding！！！  
4、Producer 发送消息给 Exchange 时，如果 Exchange 无法根据自身的类型和 RoutingKey 找对一个符合条件  
的 Queue，则消息会丢失！  
5、Broker 流控：当 Producer 发送速率过快会导致 RabbitMQ 对 Producer 进行流控，这会限制 Producer 的  
发送速率，所以，尽可能把消息发送改为异步形式，并且尽可能减少消息堆积，避免流控后拖死 Producer！  
6、单个 Broker 节点上的 Queue 尽量不要太多，太多会影响性能，RabbitMQ 采用一个 Queue 一个 进程的模  
型！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2gyNjA0Mzk2NzM5_size_16_color_FFFFFF_t_70]: /images/20220224/d0d1f94edc2a47e6aa3f3b8e7a9b1a8b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2gyNjA0Mzk2NzM5_size_16_color_FFFFFF_t_70 1]: /images/20220224/aa29a69c7195451cba7c73a12655c0ca.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2gyNjA0Mzk2NzM5_size_16_color_FFFFFF_t_70 2]: /images/20220224/4e346134ae0645bead241dc71e616041.png