RabbitMQ镜像队列集群搭建、与SpringBoot整合

时间  2021-03-13
标签 html java node web spring apache vim 浏览器 安全 springboot 栏目 RabbitMQ 繁體版
原文   原文链接

镜像模式

集群模式很是经典的就是Mirror镜像模式，保证100%数据不丢失，在实际工做中也是用的最多的，而且实现集群比较的简单。
Mirror镜像队列，目的是为了保证 RabbitMQ 数据的高可靠性解决方案，主要就是实现数据的同步，通常来说2--3个节点实现数据同步（对于100%数据可靠性解决方案通常是3节点）
感兴趣的胖友能够体验一哈新的阅读地址：http://www.zhouhong.icu/post/142  (*^▽^*)

1 前提准备

1.1 服务节点分配

服务器IP
hostname
节点说明
端口
管控台地址
192.168.2.121
zhouhong121
rabbitmq master
5672
http://192.168.2.121:15672
192.168.2.122
zhouhong122
rabbitmq slave
5672
http://192.168.2.122:15672
192.168.2.123
zhouhong123
rabbitmq slave
5672
http://192.168.2.123:15672

2 集群搭建

前提条件：修改12一、12二、123三台服务器的 hostname 而且可使用hostname 两两之间 ping 通。
  • 修改每台服务器的 hostname

vim /etc/hostname
## 修改对应的名字，好比：
zhouhong121
  • 更改每台服务器的 hosts

vim /etc/hosts
127.0.0.1   localhost localhost.localdomain localhost4 localhost4.localdomain4
::1         localhost localhost.localdomain localhost6 localhost6.localdomain6
192.168.2.121 zhouhong121
192.168.2.122 zhouhong122
192.168.2.123 zhouhong123
  •  测试，用122的hostname ping 123

2.1 集群节点安装
RabbitMQ下载：

rpm -ivh erlang-23.0.4-1.el7.x86_64.rpm
rpm -ivh socat-1.7.3.2-5.el7.lux.x86_64.rpm
rpm -ivh rabbitmq-server-3.8.5-1.el7.noarch.rpm​
 若是下载卡顿请使用我下载好的网盘进行下载便可
连接：https://pan.baidu.com/s/1diapYC19UlDy4G-4lgZWHA
提取码：jf5r
复制这段内容后打开百度网盘手机App，操做更方便哦

由于以前在121这台安装过：全部在另外两台上面快速安装便可，详细的安装请参照： 

1、安装
rpm -ivh erlang-23.0.4-1.el7.x86_64.rpm
rpm -ivh socat-1.7.3.2-5.el7.lux.x86_64.rpm
rpm -ivh rabbitmq-server-3.8.5-1.el7.noarch.rpm​
2、启动
systemctl start rabbitmq-server
3、安装web管控台
rabbitmq-plugins enable rabbitmq_management
4、添加用户
sudo rabbitmqctl add_user admin admin
sudo rabbitmqctl set_user_tags admin administrator
sudo rabbitmqctl set_permissions -p / admin "." "." ".*"  
5、重启
systemctl start rabbitmq-server
rabbitmq-plugins enable rabbitmq_management
 浏览器以admin登陆检查安装是否成功：

2.2 文件同步（注意：.erlang.cookie为隐藏文件，须要使用 -a 查看）

选择12一、12二、123任意一个节点为Master（这里选择71为Master），也就是说咱们须要把121的Cookie文件同步到12二、123节点上去，进入/var/lib/rabbitmq目录下，把/var/lib/rabbitmq/.erlang.cookie文件的权限修改成777，原来是400；而后把.erlang.cookie文件copy到各个节点下；最后把全部cookie文件权限还原为400便可。 
//进入目录修改权限；远程copy12二、123节点
cd /var/lib/rabbitmq/
chmod 777 /var/lib/rabbitmq/.erlang.cookie
scp /var/lib/rabbitmq/.erlang.cookie 192.168.2.122:/var/lib/rabbitmq/
scp /var/lib/rabbitmq/.erlang.cookie 192.168.2.123:/var/lib/rabbitmq/
// 每台服务器为默认修改权限
chmod 400 /var/lib/rabbitmq/.erlang.cookie
2.3 组成集群
2.3.1 中止服务
咱们首先中止3个节点的服务：(这里不能使用原来的命令：/etc/init.d/rabbitmq-server stop) 

rabbitmqctl stop
2.3.2 组成集群操做
接下来咱们就可使用集群命令，配置7一、7二、73为集群模式，3个节点（7一、7二、73）执行启动命令，后续启动集群使用此命令便可。 

rabbitmq-server -detached
2.3.3 slave加入集群操做（从新加入集群也是如此，以最开始的主节点为加入节点）

//注意作这个步骤的时候：须要配置/etc/hosts 必须相互可以寻址到
//在122节点上执行如下操做
rabbitmqctl stop_app
rabbitmqctl join_cluster --ram rabbit@zhouhong121
rabbitmqctl start_app
//一样在123节点上执行如下操做
rabbitmqctl stop_app
rabbitmqctl join_cluster rabbit@zhouhong121
rabbitmqctl start_app
//在另外其余节点上操做要移除的集群节点
//rabbitmqctl forget_cluster_node rabbit@zhouhong12二、12二、123
2.3.4 修改集群名称
修改集群名称（默认为第一个node名称）： 

rabbitmqctl set_cluster_name rabbitmq_cluster1
2.3.5 查看集群状态
最后在集群的任意一个节点执行命令：查看集群状态 

rabbitmqctl cluster_status

2.3.6 管控台界面（注意：这里可能以前配置的admin 角色会失效，可能须要从新配置一遍）
访问任意一个管控台节点： http://192.168.2.121:15672 如图所示

如图：121为dics 12二、123为 RAMhtml

2.4 配置镜像队列
设置镜像队列策略（在任意一个节点上执行）

rabbitmqctl set_policy ha-all "^" '{"ha-mode":"all"}'

 将全部队列设置为镜像队列，即队列会被复制到各个节点，各个节点状态一致，RabbitMQ高可用集群就已经搭建好了,咱们能够重启服务，查看其队列是否在从节点同步。java

咱们在任何一个节点上建一个队列，那这个队列将会加到其余两个节点上面

2.5 消息一致性问题
在使用rabbitmq中，消息的一致性是很是重要的一个话题。下面咱们来研究一下，在数据一致性方面，有哪些须要关注的。发送者发送消息出来，在数据一致性的要求下，咱们一般认为必须达到如下条件
  1. broker持久化消息
  2. publisher知道消息已经成功持久化
首先，咱们能够采用事务来解决此问题。每一个消息都必须经历以上两个步骤，就算一次事务成功。
事务是同步的。所以，若是采用事务，发送性能必然不好。官方给出来的性能是：

异步的方法的效率是事务方法效率的100倍。

咱们能够采用异步的方式来解决此问题。publisher发送消息后，不进行等待，而是异步监听是否成功。这种方式又分为两种模式，一种是return，另外一种是confirm. 前一种是publisher发送到exchange后，异步收到消息。第二种是publisher发送消息到exchange,queue,consumer收到消息后才会收到异步收到消息。可见，第二种方式更加安全可靠。以下所示：

可是，异步也存在些局限性。若是一旦出现broker挂机或者网络不稳定，broker已经成功接收消息，可是publisher并无收到confirm或return.这时，对于publisher来讲，只能重发消息解决问题。而在这里面，咱们会发生重复消息的问题。固然，若是业务类型要求数据一致性很是高，能够采用低效率的事务型解决方案。

3 整合SpringBoot

生产端：
  • 引入依赖
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>com.zhouhong</groupId>
  <artifactId>rabbit-producer</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.5.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <java.version>1.8</java.version>
    </properties>    
    
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency> 
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <!-- springboot rabbitmq(amqp) -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>                                    
    </dependencies>
</project>
  •  配置文件 application.properties
server.servlet.context-path=/
server.port=8011

## 镜像队列地址
spring.rabbitmq.addresses=192.168.2.121,192.168.2.122,192.168.2.123
spring.rabbitmq.username=admin
spring.rabbitmq.password=admin
## 默认虚拟主机
spring.rabbitmq.virtual-host=/
## 链接超时
spring.rabbitmq.connection-timeout=15000
## 是否使用启用消息确认模式（可靠性投递）
spring.rabbitmq.publisher-confirms=true
## 设置reture消息模式，注意要和mandatory一块儿配合使用
## spring.rabbitmq.publisher-returns=true
## spring.rabbitmq.template.mandatory=true

spring.application.name=rabbit-producer
spring.http.encoding.charset=UTF-8
spring.jackson.date-format=yyyy-MM-dd HH:mm:ss
spring.jackson.time-zone=GMT+8
spring.jackson.default-property-inclusion=NON_NULL
  •  消息发送
package com.zhouhong.rabbit.producer.component;
import java.util.Map;
import java.util.UUID;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
@Component
public class RabbbitSender {

    @Autowired
    private RabbitTemplate rabbitTemplate;
    /**
     * 这里是确认消息的回调监听接口，用于确认消息是否被 broker 所收到
     */
    final ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
        /**
         * @param CorrelationData 做为一个惟一的标识
         * @param ack broker是否落盘成功
         * @param cause 失败的一些异常信息
         */
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            // TODO Auto-generated method stub        
        }
    };

    /**
     * 对外发送消息的方法
     * @param massage 具体的消息内容
     * @param properties 额外的属性
     * @throws Exception
     */
    public void send(Object message, Map<String, Object> properties) throws Exception{
        
        MessageHeaders mhs = new MessageHeaders(properties);
        Message<?> msg = MessageBuilder.createMessage(message, mhs);
        /**
         * 使用的是confirms模式，因此在发消息以前须要监控
         */
        rabbitTemplate.setConfirmCallback(confirmCallback);
        //指定业务惟一的ID
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        
        MessagePostProcessor mpp = new MessagePostProcessor() {    
            @Override
            public org.springframework.amqp.core.Message postProcessMessage(org.springframework.amqp.core.Message message)
                    throws AmqpException {
                System.out.println("post todo: "+ message);
                return message;
            }
        };        
        rabbitTemplate.convertAndSend("exchange-1", "rabbitmq.*", msg,
                 correlationData);
        
    }
 }
 消费端：
  • 配置文件 application.properties

server.servlet.context-path=/
server.port=8012

## 镜像队列地址
spring.rabbitmq.addresses=192.168.2.121,192.168.2.122,192.168.2.123
spring.rabbitmq.username=admin
spring.rabbitmq.password=admin
## 默认虚拟主机
spring.rabbitmq.virtual-host=/
## 链接超时
spring.rabbitmq.connection-timeout=15000

## 表示消费者消息消费成功之后，须要手工的进行签收（ACK） 默认为 auto
spring.rabbitmq.listener.simple.acknowledge-mode=manual
## 线程数
spring.rabbitmq.listener.simple.concurrency=5
spring.rabbitmq.listener.simple.max-concurrency=10
## 一条一条消费
spring.rabbitmq.listener.simple.prefetch=1

##    最好不要在代码里写死配置信息，尽可能使用这种方式也就是配置文件的方式
##    在代码里使用 ${}    方式进行设置配置: ${spring.rabbitmq.listener.order.exchange.name}
## 交换机名称 
## spring.rabbitmq.listener.order.exchange.name=order-exchange
## 是否持久化 
## spring.rabbitmq.listener.order.exchange.durable=true
## type 类型
## spring.rabbitmq.listener.order.exchange.type=topic
## 规则
## spring.rabbitmq.listener.order.exchange.key=rabbitmq.*

spring.application.name=rabbit-producer
spring.http.encoding.charset=UTF-8
spring.jackson.date-format=yyyy-MM-dd HH:mm:ss
spring.jackson.time-zone=GMT+8
spring.jackson.default-property-inclusion=NON_NULL
  •  接收消息
package com.zhouhong.rabbit.consumer.component;

import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

import com.rabbitmq.client.Channel;
@Component
public class RabbbitReceive {
    /**
     * 组合使用监听
     * @param message
     * @param channel
     * @throws Exception
     */
    @RabbitListener(bindings = @QueueBinding(
                value = @Queue(value = "queue-1", durable = "true"),
                exchange = @Exchange(name = "exchange-1", 
                durable = "true", 
                type = "topic",
                ignoreDeclarationExceptions = "true"),
                key = "rabbitmq.*"
            )
            )
    @RabbitHandler
    public void onMessage(Message message, Channel channel) throws Exception {
        //一、收到消息之后进行业务端消费处理
        System.err.println("======================");
        System.err.println("消息消费：" + message.getPayload());        
        //二、处理成功以后获取deliveryTay 而且进行手工的ACK操做，由于咱们配置文件里面配置的是手工签收
        Long deliveryTay = (Long)message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
        channel.basicAck(deliveryTay, false);
    }
}
 测试：
在发送端建测试类：

 

package com.zhouhong.rabbit.producer.test;

import java.util.HashMap;
import java.util.Map;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
import com.zhouhong.rabbit.producer.component.RabbbitSender;

@RunWith(SpringRunner.class)
@SpringBootTest
public class ApplicationTest {
    @Autowired
    private RabbbitSender rabbbitSender;
    
    @Test
    public void testSender() throws Exception{
        Map<String , Object> properties = new HashMap<String, Object>();
        properties.put("key1", "你好呀，RabbitMQ!!");
        properties.put("key2", "你好呀，Kafka!!");
        rabbbitSender.send("rabbitmq-test", properties);
    
        Thread.sleep(10000);
    }
}
 一、启动消费者，观察管控台

 创建了一个咱们代码里面指定的交换机 exchange-1，而且绑定了咱们指定的队列queue-1，路由规则为 rabbitmq.*node

二、关闭消费者，只运行咱们发送端的测试方法，观察管控台

 咱们发现会有一条未消费的消息。web

三、接着，咱们再启动消费端，观察管控台

　　以前堆积的消息已经被消费掉了，因而可知，咱们的消息能够成功地投递，并且被成功地消费。
　　消费端会保证，在消息未消费以前不会将消息删掉，以此来保证消息的可靠性。通常咱们只须要作生产端跟broker的可靠性就能够了，消费端MQ会本身帮咱们作可靠性工做。
相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程