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

镜像模式

集群模式非常经典的就是Mirror镜像模式,保证100%数据不丢失,在实际工作中也是用的最多的,并且实现集群比较的简单。

Mirror镜像队列,目的是为了保证 RabbitMQ 数据的高可靠性解决方案,主要就是实现数据的同步,一般来讲2--3个节点实现数据同步(对于100%数据可靠性解决方案一般是3节点)

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 集群搭建

前提条件:修改121、122、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

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

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这台安装过:所有在另外两台上面快速安装即可,详细的安装请参照:

https://www.cnblogs.com/Tom-shushu/p/14503021.html

http://www.zhouhong.icu/post/141

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登录检查安装是否成功:

http://192.168.2.121:15672/

http://192.168.2.122:15672/

http://192.168.2.123:15672/

2.2 文件同步(注意:.erlang.cookie为隐藏文件,需要使用 -a 查看)

选择121、122、123任意一个节点为Master(这里选择71为Master),也就是说我们需要把121的Cookie文件同步到122、123节点上去,进入/var/lib/rabbitmq目录下,把/var/lib/rabbitmq/.erlang.cookie文件的权限修改为777,原来是400;然后把.erlang.cookie文件copy到各个节点下;最后把所有cookie文件权限还原为400即可。

//进入目录修改权限;远程copy122、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 组成集群操作

接下来我们就可以使用集群命令,配置71、72、73为集群模式,3个节点(71、72、73)执行启动命令,后续启动集群使用此命令即可。

rabbitmq-server -detached

2.3.3 slave加入集群操作(重新加入集群也是如此,以最开始的主节点为加入节点)

//注意做这个步骤的时候:需要配置/etc/hosts 必须相互能够寻址到

//在122节点上执行以下操作

rabbitmqctl stop_app

rabbitmqctl join_cluster --ram [email protected]

rabbitmqctl start_app

//同样在123节点上执行以下操作

rabbitmqctl stop_app

rabbitmqctl join_cluster [email protected]

rabbitmqctl start_app

//在另外其他节点上操作要移除的集群节点

//rabbitmqctl forget_cluster_node [email protected]、122、123

2.3.4 修改集群名称

修改集群名称(默认为第一个node名称):

rabbitmqctl set_cluster_name rabbitmq_cluster1

2.3.5 查看集群状态

最后在集群的任意一个节点执行命令:查看集群状态

rabbitmqctl cluster_status

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

2.3.6 管控台界面(注意:这里可能之前配置的admin 角色会失效,可能需要重新配置一遍)

访问任意一个管控台节点:http://192.168.2.121:15672 如图所示

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

如图:121为dics 122、123为 RAM

2.4 配置镜像队列

设置镜像队列策略(在任意一个节点上执行)

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

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

我们在任何一个节点上建一个队列,那这个队列将会加到其他两个节点上面

2.5 消息一致性问题

在使用rabbitmq中,消息的一致性是非常重要的一个话题。下面我们来研究一下,在数据一致性方面,有哪些需要关注的。发送者发送消息出来,在数据一致性的要求下,我们通常认为必须达到以下条件

  1. broker持久化消息
  2. publisher知道消息已经成功持久化

首先,我们可以采用事务来解决此问题。每个消息都必须经历以上两个步骤,就算一次事务成功。

事务是同步的。因此,如果采用事务,发送性能必然很差。官方给出来的性能是:

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

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

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

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

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

3 整合SpringBoot

本篇文章所有代码请自行到我的GitHub上拉取:https://github.com/Tom-shushu/Distributed-system-learning-notes

生产端:

  • 引入依赖

<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 {

//1、收到消息以后进行业务端消费处理

System.err.println("======================");

System.err.println("消息消费:" + message.getPayload());

//2、处理成功之后获取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);

}

}

 1、启动消费者,观察管控台

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

建立了一个我们代码里面指定的交换机 exchange-1,并且绑定了我们指定的队列queue-1,路由规则为 rabbitmq.*

2、关闭消费者,只运行我们发送端的测试方法,观察管控台

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

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

3、接着,我们再启动消费端,观察管控台

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

  之前堆积的消息已经被消费掉了,由此可见,我们的消息可以成功地投递,而且被成功地消费。

  消费端会保证,在消息未消费之前不会将消息删掉,以此来保证消息的可靠性。一般我们只需要做生产端跟broker的可靠性就可以了,消费端MQ会自己帮我们做可靠性工作。

以上是 SpringBoot整合RabbitMQ镜像队列集群搭建 的全部内容, 来源链接: www.h5w3.com/122528.html

回到顶部