欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

阶段七模块二 RabbitMQ

程序员文章站 2022-07-10 20:22:20
...

内容输出来源:拉钩教育Java就业训练营

面试题:避免消息堆积

  1. workqueue,多个消费者监听同一个队列
  2. 接收到消息后,通过线程池,异步消费

1 RabbitMQ

1.1 MQ(Message Queue)消息队列

消息队列中间件,是分布式系统中的重要组件
主要解决,异步处理,应用解耦,流量削峰等问题
从而实现高性能,高可用,可伸缩和最终一致性的架构

1.1.1 异步处理

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-7BLezibG-1618660866150)(RabbitMQ.assets/RabbitMQ详解.jpg)]

1.1.2 应用解耦

订单系统需要通知库存系统,如果库存系统异常,则订单调用库存失败,导致下单失败

订单系统和库存系统耦合度太高

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-6hEK1Xvs-1618660866152)(RabbitMQ.assets/RabbitMQ详解-1618448182873.jpg)]

下单的时候,库存系统不能正常运行,也不会影响下单,因为下单后,订单系统写入消息队列就不再关心其他的后续操作了,实现了订单系统和库存系统的应用解耦;

1.1.3 流量削峰

抢购,秒杀等业务,针对高并发的场景
因为流量过大,暴增会导致应用挂掉,为解决这个问题,在前端加入消息队列

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-lzNpU3SR-1618660866153)(RabbitMQ.assets/RabbitMQ详解-1618448372074.jpg)]

1.2 RabbitMQ各组件

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-TQMV618e-1618660866155)(RabbitMQ.assets/RabbitMQ详解-1618448854663.jpg)]

  • Broker:消息队列服务器实体
  • Virtual Host:虚拟主机
    • 标识一批交换机、消息队列和相关对象,形成的整体
    • 虚拟主机是共享相同的身份认证和加密环境的独立服务器域
    • 每个vhost本质上就是一个mini版的RabbitMQ服务器,拥有自己的队列、交换器、绑定和权限机制
    • vhost是AMQP概念的基础,RabbitMQ默认的vhost是 /,必须在链接时指定
  • Exchange:交换器(路由)
    • 用来接收生产者发送的消息并将这些消息路由给服务器中的队列
  • Queue:消息队列
    • 用来保存消息直到发送给消费者。
    • 它是消息的容器,也是消息的终点。
    • 一个消息可投入一个或多个队列。
    • 消息一直在队列里面,等待消费者连接到这个队列将其取走。
  • Banding:绑定,用于消息队列和交换机之间的关联。
  • Channel:通道(信道)
    • 多路复用连接中的一条独立的双向数据流通道。
    • 信道是建立在真实的TCP连接内的 虚拟链接
    • AMQP命令都是通过信道发出去的,不管是发布消息、订阅队列还是接收消息,都是通过信道完成的
    • 因为对于操作系统来说,建立和销毁TCP连接都是非常昂贵的开销,所以引入了信道的概念,用来复用TCP连接。
  • Connection:网络连接,比如一个TCP连接。
  • Publisher:消息的生产者,也是一个向交换器发布消息的客户端应用程序。
  • Consumer:消息的消费者,表示一个从消息队列中取得消息的客户端应用程序。
  • Message:消息
    • 消息是不具名的,它是由消息头和消息体组成。
    • 消息体是不透明的,而消息头则是由一系列的可选属性组成,这些属性包括routing-key(路由键)、priority(优先级)、delivery-mode(消息可能需要持久性存储[消息的路由模式])等。

2 RabbitMQ安装及使用

2.1 安装及启动

2.1.1 安装
rpm -ivh erlang-21.3.8.16-1.el7.x86_64.rpm
rpm -ivh socat-1.7.3.2-5.el7.lux.x86_64.rpm
rpm -ivh rabbitmq-server-3.8.6-1.el7.noarch.rpm
2.1.2 启动后台管理插件
rabbitmq-plugins enable rabbitmq_management
2.1.3 启动RabbitMq
systemctl start rabbitmq-server.service
systemctl status rabbitmq-server.service
systemctl restart rabbitmq-server.service
systemctl stop rabbitmq-server.service
2.1.4 查看进程
ps -ef | grep rabbitmq
2.1.5 测试

关闭防火墙

systemctl stop firewalld

浏览器输入:http://ip:15672

1.创建账号

rabbitmqctl add_user laosun 123456

2.设置用户角色

rabbitmqctl set_user_tags laosun administrator

3.设置用户权限

rabbitmqctl set_permissions -p "/" laosun ".*"
".*" ".*"

4.查看当前用户和角色

rabbitmqctl list_users

5.更改密码

rabbitmqctl change_password wx 123123

管理界面端口:

5672:RabbitMQ提供给编程语言客户端链接的端口
15672:RabbitMQ管理界面的端口
25672:RabbitMQ集群的端口

2.2 java与RabbitMQ连接

<dependencies>
    <dependency>
        <groupId>com.rabbitmq</groupId>
        <artifactId>amqp-client</artifactId>
        <version>5.7.3</version>
	</dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-log4j12</artifactId>
        <version>1.7.25</version>
        <scope>compile</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.commons</groupId>
        <artifactId>commons-lang3</artifactId>
        <version>3.9</version>
    </dependency>
</dependencies>

日志依赖log4j

log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.Target=System.out
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %m%n
log4j.appender.file=org.apache.log4j.FileAppender
log4j.appender.file.File=rebbitmq.log
log4j.appender.file.layout=org.apache.log4j.PatternLayout
log4j.appender.file.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %l %m%n
log4j.rootLogger=debug, stdout,file
package util;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class ConnectionUtil {
    public static Connection getConnection() throws Exception{
        //1.创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //2.在工厂对象中设置MQ的连接信息(ip,port,vhost,username,password)
        factory.setHost("192.168.163.128");
        factory.setPort(5672);
        factory.setVirtualHost("/lagou");
        factory.setUsername("wx");
        factory.setPassword("123456");
        //3.通过工厂获得与MQ的连接
        Connection connection = factory.newConnection();
        return connection;
    }
    public static void main(String[] args) throws Exception{
        Connection connection = getConnection();
        System.out.println("connection = " + connection);
        connection.close();
    }
}

2.3 RabbitMQ模式

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-lvs5Pd2c-1618660866156)(RabbitMQ.assets/RabbitMQ详解-1618451692977.jpg)]

5种消息模型,大体分为两类:
1和2属于点对点
3、4、5属于发布订阅模式(一对多)

点对点模式:P2P(point to point)模式包含三个角色:

消息队列(queue),发送者(sender),接收者(receiver)

每个消息发送到一个特定的队列中,接收者从中获得消息
队列中保留这些消息,直到他们被消费或超时

特点:

每个消息只有一个消费者,一旦消费,消息就不在队列中了

发送者和接收者之间没有依赖性,发送者发送完成,不管接收者是否运行,都不会影响消息发送到队列中

接收者成功接收消息之后需向对象应答成功

发布订阅模式:publish(Pub)/subscribe(Sub)

pub/sub模式包含三个角色:交换机(exchange),发布者(publisher),订阅者(subcriber)

多个发布者将消息发送交换机,系统将这些消息传递给多个订阅者

特点:

每个消息可以有多个订阅者

发布者和订阅者之间在时间上有依赖,对于某个交换机的订阅者,必须创建一个订阅后,才能消费发布者的消息

为了消费消息,订阅者必须保持运行状态

2.3.1 简单模式

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-u65S4aFv-1618660866158)(RabbitMQ.assets/RabbitMQ详解-1618453698752.jpg)]

生产者

public class Sender {
    public static void main(String[] args) throws Exception {
        String msg = "wx:Hello,RabbitMQ!";
        // 1.获得连接
        Connection connection = ConnectionUtil.getConnection();
        // 2.在连接中创建通道(信道)
        Channel channel = connection.createChannel();
        // 3.创建消息队列(1,2,3,4,5)
        /*
        参数1:队列的名称
        参数2:队列中的数据是否持久化
        参数3:是否排外(是否支持扩展,当前队列只能自己用,不能给别人用)
        参数4:是否自动删除(当队列的连接数为0时,队列会销毁,不管队列是否还存保存数据)
        参数5:队列参数(没有参数为null)
        */
        channel.queueDeclare("queue1",false,false,false,null);
        // 4.向指定的队列发送消息(1,2,3,4)
        /*
        参数1:交换机名称,当前是简单模式,也就是P2P模式,没有交换机,所以名称为""
        参数2:目标队列的名称
        参数3:设置消息的属性(没有属性则为null)
        参数4:消息的内容(只接收字节数组)
        */
        channel.basicPublish("","queue1",null,msg.getBytes());
        System.out.println("发送:" + msg);
        // 5.释放资源
        channel.close();
        connection.close();
    }
}

消费者

public class Recer {
    public static void main(String[] args) throws Exception {
    // 1.获得连接
    Connection connection = ConnectionUtil.getConnection();
    // 2.获得通道(信道)
    Channel channel = connection.createChannel();
    // 3.从信道中获得消息
    DefaultConsumer consumer = new DefaultConsumer(channel){
        @Override //交付处理(收件人信息,包裹上的快递标签,协议的配置,消息)
        public void handleDelivery(String consumerTag, Envelope envelope,
        AMQP.BasicProperties properties, byte[] body) throws IOException {
            // body就是从队列中获取的消息
            String s = new String(body);
            System.out.println("接收 = " + s);
        }
    };
        // 4.监听队列 true:自动消息确认
        channel.basicConsume("queue1", true,consumer);
    }
}

消息确认机制ACK

消息一旦被消费,消息就会立刻从队列中移除

如果消费者接收消息后,还没执行操作就抛异常宕机导致消费失败,但是RabbitMQ无从得知,这样消息就丢失了

RabbitMQ有一个ACK机制,当消费者获取消息后,会向RabbitMQ发送回执ACK,告知消息已经被接收

回执ACK分为两种情况:

自动ACK:消息接收后,消费者立刻自动发送ACK(快递放在快递柜)
手动ACK:消息接收后,不会发送ACK,需要手动调用(快递必须本人签收)

如果消息不太重要,丢失也没有影响,自动ACK会比较方便

如果消息非常重要,最好消费完成手动ACK,如果自动ACK消费后,RabbitMQ就会把消息从队列中删除,如果此时消费者抛异常宕机,那么消息就永久丢失了

// false:手动消息确认
channel.basicConsume("queue1", false, consumer);

实例:

public class RecerByACK {
    public static void main(String[] args) throws Exception {
        // 1.获得连接
        Connection connection = ConnectionUtil.getConnection();
        // 2.获得通道(信道)
        final Channel channel = connection.createChannel();
        // 3.从信道中获得消息
        DefaultConsumer consumer = new DefaultConsumer(channel){
            @Override //交付处理(收件人信息,包裹上的快递标签,协议的配置,消息)
            public void handleDelivery(String consumerTag, Envelope envelope,
            AMQP.BasicProperties properties, byte[] body) throws IOException {
                // body就是从队列中获取的消息
                String s = new String(body);
                System.out.println("接收 = " + s);
                // 手动确认(收件人信息,是否同时确认多个消息)
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        // 4.监听队列,false:手动消息确认
        channel.basicConsume("queue1", false,consumer);
    }
}
2.3.2 工作队列模式

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-YOR9Kv0b-1618660866158)(RabbitMQ.assets/RabbitMQ详解-1618453725057.jpg)]

当我们运行许多消费者程序时,消息队列中的任务会被众多消费者共
享,但其中某一个消息只会被一个消费者获取

生产者

public class MessageSender {
    public static void main(String[] args) throws Exception{
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        // 声明队列(此处为生产者,创建队列)注明出餐口位置,通知大家来排队
        channel.queueDeclare("test_work_queue",false,false,false,null);
        for(int i = 1;i<=100;i++) {
            String msg = "羊肉串 --> "+i;
            channel.basicPublish("", "test_work_queue", null, msg.getBytes());
            System.out.println("师傅烤好:" + msg);
        }
        channel.close();
        connection.close();
    }
}

1号消费者

public class MessageReceiver1 {
    static int i = 1; // 记录执行次数
    public static void main(String[] args) throws IOException, TimeoutException{
        Connection connection = ConnectionUtil.getConnection();
        final Channel channel = connection.createChannel();
        // 声明队列(此处为消费者,不是声明创建队列,而是获取,二者代码相同)出餐口排队
        channel.queueDeclare("test_work_queue",false,false,false,null);
        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
            AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body);
                System.out.println("【顾客1】吃掉 " + msg+" ! 共吃【"+i++ +"】串");
                // 撸一会,有延迟
                try {
                	Thread.sleep(200);
                }catch (InterruptedException e){
                }
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        channel.basicConsume("test_work_queue", false, consumer);
    }
}

2号消费者

public class MessageReceiver2 {
    static int i = 1; // 记录执行次数
    public static void main(String[] args) throws IOException, TimeoutException{
        Connection connection = ConnectionUtil.getConnection();
        final Channel channel = connection.createChannel();
        // 声明队列(此处为消费者,不是声明创建队列,而且获取,二者代码相同)出餐口排队
        channel.queueDeclare("test_work_queue",false,false,false,null);
        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
            AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body);
                System.out.println("【顾客2】吃掉 " + msg+" ! 共吃【"+i++ +"】串");
                // 撸一会,有延迟
                try {
                	Thread.sleep(200);
                }catch (InterruptedException e){
                }
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        channel.basicConsume("test_work_queue", false, consumer);
    }
}

先运行1、2号消费者,在运行生产者

两个消费者的消费速度(线程休眠时间)不一致,但消费数量相同,各消费50个。

解决这一问题

// 声明队列(此处为消费者,不是声明创建队列,而且获取,二者代码相同)出餐口排队
channel.queueDeclare("test_work_queue",false,false,false,null);

在1、2号消费者中上述声明队列代码之后添加

// 可以理解为:快递一个一个送,送完一个再送下一个,速度快的送件就多
channel.basicQos(1);
2.3.3 发布订阅模式

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-uP6DNuWa-1618660866159)(RabbitMQ.assets/RabbitMQ详解-1618455117555.jpg)]

P生产者发送信息给X路由,X将信息转发给绑定X的队列

X队列将信息通过信道发送给消费者,从而进行消费

必须先创建路由

路由在生产者程序中创建

因为路由没有存储消息的能力,当生产者将信息发送给路由后,消费者还没有运行,所以没有队列,路由并不知道将信息发送给谁

生产者

public class Sender {
    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        // 声明路由(路由名,路由类型)
        // fanout:不处理路由键(只需要将队列绑定到路由上,发送到路由的消息都会被转发到与该路由绑定的所有队列上)
        channel.exchangeDeclare("test_exchange_fanout", "fanout");
        String msg = "hello,大家好!";
        channel.basicPublish("test_exchange_fanout", "", null, msg.getBytes());
        System.out.println("生产者:" + msg);
        channel.close();
        connection.close();
    }
}

1号消费者

public class Recer1 {
    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare("test_exchange_fanout_queue_1",false,false,false,null);
        // 绑定路由(关注)
        /*
        参数1:队列名
        参数2:交换器名称
        参数3:路由key(暂时无用,""即可)
        */
        channel.queueBind("test_exchange_fanout_queue_1",
        "test_exchange_fanout", "");
        DefaultConsumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
            AMQP.BasicProperties properties, byte[] body) throws IOException {
                String s = new String(body);
                System.out.println("【消费者1】 = " + s);
            }
        };
        // 4.监听队列 true:自动消息确认
        channel.basicConsume("test_exchange_fanout_queue_1", true,consumer);
    }
}

2号消费者同1号代码一致,但需要更改队列名称为2.

2.3.4 路由模式

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-4CVJrXbk-1618660866159)(RabbitMQ.assets/RabbitMQ详解-1618456013569.jpg)]

路由会根据类型进行定向分发消息给不同的队列,如图所示

生产者

public class Sender {
    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        // 声明路由(路由名,路由类型)
        // direct:根据路由键进行定向分发消息
        channel.exchangeDeclare("test_exchange_direct", "direct");
        String msg = "用户注册,【userid=S101】";
        channel.basicPublish("test_exchange_direct", "insert", null,msg.getBytes());
        System.out.println("[用户系统]:" + msg);
        channel.close();
        connection.close();
    }
}

1号消费者

public class Recer1 {
    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare("test_exchange_direct_queue_1",false,false,false,null);
        // 绑定路由(如果路由键的类型是 添加,删除,修改 的话,绑定到这个队列1上)
        channel.queueBind("test_exchange_direct_queue_1",
        "test_exchange_direct", "insert");
        channel.queueBind("test_exchange_direct_queue_1",
        "test_exchange_direct", "update");
        channel.queueBind("test_exchange_direct_queue_1",
        "test_exchange_direct", "delete");
        DefaultConsumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
            AMQP.BasicProperties properties, byte[] body) throws IOException {
                String s = new String(body);
                System.out.println("【消费者1】 = " + s);
            }
        };
        // 4.监听队列 true:自动消息确认
        channel.basicConsume("test_exchange_direct_queue_1", true,consumer);
    }
}

2号消费者

public class Recer2 {
    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare("test_exchange_direct_queue_2",false,false,false,null);
        // 绑定路由(如果路由键的类型是 查询 的话,绑定到这个队列2上)
        channel.queueBind("test_exchange_direct_queue_2",
        "test_exchange_direct", "select");
        DefaultConsumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
            AMQP.BasicProperties properties, byte[] body) throws IOException {
                String s = new String(body);
                System.out.println("【消费者2】 = " + s);
            }
        };
        // 4.监听队列 true:自动消息确认
        channel.basicConsume("test_exchange_direct_queue_2", true,consumer);
    }
}
  1. 记住运行程序的顺序,先运行一次sender(创建路由器)
  2. 有了路由器之后,在创建两个Recer1和Recer2,进行队列绑定
  3. 再次运行sender,发出消息
2.3.5 通配符模式

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ZHH6sTdY-1618660866160)(RabbitMQ.assets/RabbitMQ详解-1618456698593.jpg)]

和路由模式基本一样,区别在于,路由键支持模糊查询

*:只能匹配一个词(正好一个词,多一个不行,少一个也不行)
#:匹配0个或更多个词

生产者

public class Sender {
    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        // 声明路由(路由名,路由类型)
        // topic:模糊匹配的定向分发
        channel.exchangeDeclare("test_exchange_topic", "topic");
        String msg = "商品降价";
        channel.basicPublish("test_exchange_topic", "product.price", null,
        msg.getBytes());
        System.out.println("[用户系统]:" + msg);
        channel.close();
        connection.close();
    }
}

1号消费者

public class Recer1 {
    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare("test_exchange_topic_queue_1",false,false,false,null);
        // 绑定路由(绑定 用户相关 的消息)
        channel.queueBind("test_exchange_topic_queue_1", "test_exchange_topic","user.#");
        DefaultConsumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
            AMQP.BasicProperties properties, byte[] body) throws IOException {
                String s = new String(body);
                System.out.println("【消费者1】 = " + s);
            }
        };
        // 4.监听队列 true:自动消息确认
        channel.basicConsume("test_exchange_topic_queue_1", true,consumer);
    }
}

2号消费者

public class Recer2 {
    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare("test_exchange_topic_queue_2",false,false,false,null);
        // 绑定路由(绑定 商品和订单相关 的消息)
        channel.queueBind("test_exchange_topic_queue_2", "test_exchange_topic","product.#");
        channel.queueBind("test_exchange_topic_queue_2", "test_exchange_topic","order.#");
        DefaultConsumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
            AMQP.BasicProperties properties, byte[] body) throws IOException {
                String s = new String(body);
                System.out.println("【消费者2】 = " + s);
            }
        };
        // 4.监听队列 true:自动消息确认
        channel.basicConsume("test_exchange_topic_queue_2", true,consumer);
    }
}

2.4 持久化

想要将消息持久化,那么 路由和队列都要持久化 才可以

生产者

public class Sender {
    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        // 声明路由(路由名,路由类型,持久化)
        channel.exchangeDeclare("test_exchange_topic", "topic",true);
        String msg = "商品降价";
        // 发送消息(第三个参数作用是让消息持久化)
        channel.basicPublish("test_exchange_topic", "product.price",
        MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes());
        System.out.println("[用户系统]:" + msg);
        channel.close();
        connection.close();
    }
}

消费者

public class Recer1 {
    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        // 声明队列( 第二个参数为true:支持持久化)
        channel.queueDeclare("test_exchange_topic_queue_1",true,false,false,null);
        channel.queueBind("test_exchange_topic_queue_1", "test_exchange_topic",
        "user.#");
        DefaultConsumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
            AMQP.BasicProperties properties, byte[] body) throws IOException {
                String s = new String(body);
                System.out.println("【消费者1】 = " + s);
            }
        };
        channel.basicConsume("test_exchange_topic_queue_1", true,consumer);
    }
}

2.5 Spring整合RabbitMQ

五种消息模型,在企业中应用最广泛的就是最后一种:定向匹配topic

生产端工程

pom.xml

<dependency>
    <groupId>org.springframework.amqp</groupId>
    <artifactId>spring-rabbit</artifactId>
    <version>2.0.1.RELEASE</version>
</dependency>
<dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-log4j12</artifactId>
    <version>1.7.25</version>
    <scope>compile</scope>
</dependency>
<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-lang3</artifactId>
    <version>3.9</version>
</dependency>

spring-rabbitmq-producer.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:rabbit="http://www.springframework.org/schema/rabbit"
    xsi:schemaLocation="
    http://www.springframework.org/schema/beans
    http://www.springframework.org/schema/beans/spring-beans.xsd
    http://www.springframework.org/schema/rabbit
    http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">
    <!-- 1.配置连接 -->
    <rabbit:connection-factory
    id="connectionFactory"
    host="192.168.163.128"
    port="5672"
    username="wx"
    password="123456"
    virtual-host="/lagou"
    />
    <!-- 2.配置队列 -->
    <rabbit:queue name="test_spring_queue_1"/>
    <!-- 3.配置rabbitAdmin:主要用于在Java代码中对理队和队列进行管理,用于创建、绑定、删
    除队列与交换机,发送消息等 -->
    <rabbit:admin connection-factory="connectionFactory"/>
    <!-- 4.配置topic类型exchange;队列绑定到交换机 -->
    <rabbit:topic-exchange name="spring_topic_exchange">
    <rabbit:bindings>
    <rabbit:binding queue="test_spring_queue_1" pattern="msg.#"/>
    </rabbit:bindings>
    </rabbit:topic-exchange>
    <!-- 5. 配置消息对象json转换类 -->
    <bean id="jsonMessageConverter"
    class="org.springframework.amqp.support.converter.Jackson2JsonMessageConverter"/
    >
    <!-- 6. 配置RabbitTemplate(消息生产者) -->
    <rabbit:template id="rabbitTemplate"
    connection-factory="connectionFactory"
    exchange="spring_topic_exchange"
    message-converter="jsonMessageConverter"
    />
</beans>

发送消息

public class Sender {
    public static void main(String[] args) {
        // 1.创建spring容器
        ClassPathXmlApplicationContext context = new
        ClassPathXmlApplicationContext("spring/spring-rabbitmq-producer.xml");
        // 2.从容器中获取对象
        RabbitTemplate template = context.getBean(RabbitTemplate.class);
        // 3.发送消息
        Map<String, String> map = new HashMap();
        map.put("name", "大佬孙");
        map.put("email", "[email protected]");
        template.convertAndSend("msg.user", map);
        context.close();
    }
}

消费端工程

pox.xml同生产端一致

spring-rabbitmq-consumer.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:rabbit="http://www.springframework.org/schema/rabbit"
    xmlns:context="http://www.springframework.org/schema/context"
    xsi:schemaLocation="
    http://www.springframework.org/schema/beans
    http://www.springframework.org/schema/beans/spring-beans.xsd
    http://www.springframework.org/schema/rabbit
    http://www.springframework.org/schema/rabbit/spring-rabbit.xsd
    http://www.springframework.org/schema/context
    http://www.springframework.org/schema/context/spring-context.xsd">
    <!-- 1. 配置连接 -->
    <rabbit:connection-factory
    id="connectionFactory"
    host="192.168.163.128"
    port="5672"
    username="wx"
    password="123456"
    virtual-host="/lagou"
    />
    <!-- 2. 配置队列 -->
    <rabbit:queue name="test_spring_queue_1"/>
    <!-- 3.配置rabbitAdmin -->
    <rabbit:admin connection-factory="connectionFactory"/>
    <!-- 4.springIOC注解扫描包-->
    <context:component-scan base-package="listener"/>
    <!-- 5.配置监听 -->
    <rabbit:listener-container connection-factory="connectionFactory">
    	<rabbit:listener ref="consumerListener" queuenames="test_spring_queue_1" />
    </rabbit:listener-container>
</beans>

MessageListener接口用于spring容器接收到消息后处理消息

如果需要使用自己定义的类型 来实现 处理消息时,必须实现该接口,并重写onMessage()方法

当spring容器接收消息后,会自动交由onMessage进行处理

消费者

@Component
public class ConsumerListener implements MessageListener {
    // jackson提供序列化和反序列中使用最多的类,用来转换json的
    private static final ObjectMapper MAPPER = new ObjectMapper();
        public void onMessage(Message message) {
        try {
            // 将message对象转换成json
            JsonNode jsonNode = MAPPER.readTree(message.getBody());
            String name = jsonNode.get("name").asText();
            String email = jsonNode.get("email").asText();
            System.out.println("从队列中获取:【"+name+"的邮箱是:"+email+"】");
        } catch (Exception e){
        	e.printStackTrace();
        }
    }
}

启动项目

public class TestRunner {
    public static void main(String[] args) throws Exception {
        // 获得容器
        ClassPathXmlApplicationContext context = new
        ClassPathXmlApplicationContext("spring/spring-rabbitmq-consumer.xml");
        // 让程序一直运行,别终止
        System.in.read();
    }
}

2.6 消息成功确认机制

在发送的消息出现错误是,我们无法保证消息能够成功投递。理想情况下应为消息发送后出错则将出错前的消息返回,并不发送,此时需要考虑用事务机制。

2.6.1 事务机制

AMQP协议提供的一种保证消息成功投递的方式,通过信道开启 transactional 模式

并利用信道 的三个方法来实现以事务方式 发送消息,若发送失败,通过异常处理回滚事务,确保消息成功投递
channel.txSelect(): 开启事务
channel.txCommit() :提交事务
channel.txRollback() :回滚事务

生产者

public class Sender {
    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare("test_transaction", "topic");
        channel.txSelect(); // 开启事务
        try {
            channel.basicPublish("test_transaction", "product.price", null, "商品
            降价1".getBytes());
            // System.out.println(1/0); // 模拟异常!
            channel.basicPublish("test_transaction", "product.price", null, "商品
            降价2".getBytes());
            System.out.println("消息全部发出!");
            channel.txCommit(); // 事务提交
        }catch (Exception e){
            System.out.println("由于系统异常,消息全部撤回!");
            channel.txRollback(); // 事务回滚
        	e.printStackTrace();
        }finally {
            channel.close();
            connection.close();
        }
    }
}

消费者

public class Recer {
    public static void main(String[] args) throws Exception {
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare("test_transaction_queue",false,false,false,null);
        channel.queueBind("test_transaction_queue", "test_transaction",
        "product.#");
        DefaultConsumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
            AMQP.BasicProperties properties, byte[] body) throws IOException {
                String s = new String(body);
                System.out.println("【消费者】 = " + s);
            }
        };
        channel.basicConsume("test_transaction_queue", true,consumer);
    }
}
2.6.2 Confirm发布确认机制

RabbitMQ为了保证消息的成功投递,采用通过AMQP协议层面为我们提供事务机制的方案,但是采用事务会大大降低消息的吞吐量

事务效率为什么会这么低呢?试想一下:10条消息,前9条成功,如果第10条失败,那么9条消息要全部撤销回滚。太太太浪费
而confirm模式则采用补发第10条的措施来完成10条消息的送达

spring-rabbitmq-producer.xml

在前文中的xml文件中进行修改,添加生产者确认机制

<!--1.配置连接,启动生产者确认机制: publisher-confirms="true"-->
<rabbit:connection-factory id="connectionFactory"
                            host="192.168.163.128"
                            port="5672"
                            username="wx"
                            password="123456"
                            virtual-host="/lagou"
                            publisher-confirms="true"
/>
<!--6.配置rabbitmq的模版,添加确认回调处理类:confirmcallback="
msgSendConfirmCallback"-->
<rabbit:template id="rabbitTemplate"
                connection-factory="connectionFactory"
                exchange="spring_topic_exchange"
                message-converter="jsonMessageConverter"
                confirm-callback="msgSendConfirmCallback"/>
<!--7.确认机制处理类-->
<bean id="msgSendConfirmCallback" class="confirm.MsgSendConfirmCallback"/>

消息确认处理类

判断消息是否能够成功投递的类

/**
* @BelongsProject: spring-rabbitmq-producer
* @Author: GuoAn.Sun
* @Description: 确认机制
*/
@Component
public class MsgSendConfirmCallback implements
RabbitTemplate.ConfirmCallback {
    public void confirm(CorrelationData correlationData, boolean b, String s) {
        if (b){
        	System.out.println("消息确认成功!!");
        } else {
            System.out.println("消息确认失败。。。");
            // 如果本条消息一定要发送到队列中,例如下订单消息,我们可以采用消息补发
            // 采用递归(固定次数,不可无限)或 redis+定时任务
        }
    }
}

log4j.properties

能够打印程序运行的日志(可查询错误信息),但是否存在不影响消息传递的功能实现。

log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.Target=System.out
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %m%n
log4j.appender.file=org.apache.log4j.FileAppender
log4j.appender.file.File=rabbitmq.log
log4j.appender.file.layout=org.apache.log4j.PatternLayout
log4j.appender.file.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %l%m%n
log4j.rootLogger=debug, stdout,file

发送消息

public class Sender {
    public static void main(String[] args) {
        ClassPathXmlApplicationContext context = new
        ClassPathXmlApplicationContext("spring/spring-rabbitmq-producer.xml");
        RabbitTemplate rabbitTemplate =
        context.getBean(RabbitTemplate.class);
        Map<String,String> map = new HashMap<String, String>();
        map.put("name", "吕布");
        map.put("email", "[email protected]");
        // 第一个参数是路由名称,
        // 不写,则使用spring容器中创建的路由
        // 乱写一个,因为路由名错误导致报错,则进入消息确认失败流程
        rabbitTemplate.convertAndSend("x","msg.user",map);
        System.out.println("ok");
        context.close();
    }
}

2.7 消费端限流

当数据量特别大的时候,我们对生产端限流肯定是不科学的,因为有时候并发量就是特别大,有时候并发量又特别少,这是用户的行为,我们是无法约束的

所以我们应该对消费端限流,用于保持消费端的稳定

RabbitMQ 提供了一种 Qos (Quality of Service,服务质量)服务质量保证功能
即在非自动确认消息的前提下,如果一定数目的消息未被确认前,不再进行消费新的消息

消费者进行限流处理

<!--5.配置监听-->
<!-- prefetch="3" 一次性消费的消息数量。会告诉 RabbitMQ 不要同时给一个消费者推送多于
N 个消息,一旦有 N 个消息还没有ack,则该 consumer 将阻塞,直到消息被ack-->
<!-- acknowledge-mode: manual 手动确认-->
<rabbit:listener-container connection-factory="connectionFactory"
prefetch="3" acknowledge="manual">
	<rabbit:listener ref="consumerListener" queuenames="test_spring_queue_1" />
</rabbit:listener-container>
// AbstractAdaptableMessageListener用于在spring容器接收到消息后用于处理消息的抽象基类
@Component
public class ConsumerListener extends AbstractAdaptableMessageListener {
    // jackson提供序列化和反序列中使用最多的类,用来转换json的
    private static final ObjectMapper MAPPER = new ObjectMapper();
    public void onMessage(Message message, Channel channel) throws Exception
    {
        try {
            // String str = new String(message.getBody());
            // 将message对象转换成json
            JsonNode jsonNode = MAPPER.readTree(message.getBody());
            String name = jsonNode.get("name").asText();
            String email = jsonNode.get("email").asText();
            System.out.println("从队列中获取:【"+name+"的邮箱是:"+email+"】");
            long deliveryTag =
            message.getMessageProperties().getDeliveryTag();
            //确认收到(参数1,参数2)
            /*
            参数1:RabbitMQ 向该 Channel 投递的这条消息的唯一标识 ID,是一个单调递增的正整数,delivery_tag 的范围仅限于 Channel
            参数2:为了减少网络流量,手动确认可以被批处理,当该参数为 true 时,则可以一次性确认 delivery_tag 小于等于传入值的所有消息
            */
            channel.basicAck(deliveryTag , true);
            Thread.sleep(3000);
            System.out.println("休息三秒然后再接收消息");
        } catch (Exception e){
        	e.printStackTrace();
        }
    }
}

2.8 过期时间TTL

Time To Live:生存时间、还能活多久,单位毫秒

在这个周期内,消息可以被消费者正常消费,超过这个时间,则自动删除(其实是被称为deadmessage并投入到死信队列,无法消费该消息)

RabbitMQ可以对消息队列设置TTL

通过队列设置,队列中所有消息都有相同的过期时间
对消息单独设置,每条消息的TTL可以不同(更颗粒化)

2.8.1 设置队列TTL

spring-rabbitmq-producer.xml

5s后自动删除

<!--2.重新配置一个队列,同时,对队列中的消息设置过期时间-->
<rabbit:queue name="test_spring_queue_ttl" auto-declare="true">
    <rabbit:queue-arguments>
    	<entry key="x-message-ttl" value-type="long" value="5000"></entry>
    </rabbit:queue-arguments>
</rabbit:queue>
2.8.2 设置消息TTL

设置某条消息的ttl,只需要在创建发送消息时指定即可

<!--2.配置队列-->
<rabbit:queue name="test_spring_queue_ttl_2">
/**
* @BelongsProject: spring-rabbitmq-producer
* @Author: GuoAn.Sun
* @Description: 生产者
*/
public class Sender2 {
    public static void main(String[] args) {
        ClassPathXmlApplicationContext context = new
        ClassPathXmlApplicationContext("spring/spring-rabbitmq-producer.xml");
        RabbitTemplate rabbitTemplate = context.getBean(RabbitTemplate.class);
        // 创建消息配置对象
        MessageProperties messageProperties = new MessageProperties();
        // 设置消息过期时间
        messageProperties.setExpiration("6000");
        // 创建消息
        Message message = new Message("6秒后自动删除".getBytes(),
        messageProperties);
        // 发送消息
        rabbitTemplate.convertAndSend("msg.user", message);
        System.out.println("消息已发出...");
        context.close();
    }
}

如果同时设置了queue和message的TTL值,则二者中较小的才会起作用

2.9 死信队列

DLX(Dead Letter Exchanges)死信交换机/死信邮箱,当消息在队列中由于某些原因没有被及时消费而变成死信(dead message)后,这些消息就会被分发到DLX交换机中,而绑定DLX交换机的队列,称之为:“死信队列”

消息没有被及时消费的原因:

消息被拒绝(basic.reject/ basic.nack)并且不再重新投递 requeue=false
消息超时未消费达到最大队列长度

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-xSZkAxiL-1618660866161)(RabbitMQ.assets/RabbitMQ详解-1618546937827.jpg)]

spring-rabbitmq-producer-dlx.xml

<!--1.配置连接-->
<rabbit:connection-factory id="connectionFactory"
                            host="192.168.163.128"
                            port="5672"
                            username="wx"
                            password="123456"
                            virtual-host="/lagou"/>
<!--3.配置rabbitAdmin:主要用于在java代码中对队列的管理,用来创建,绑定,删除队列与交
换机,发送消息等-->
<rabbit:admin connection-factory="connectionFactory"/>
<!--6.配置rabbitmq的模版-->
<rabbit:template id="rabbitTemplate"
                 connection-factory="connectionFactory"
					exchange="my_exchange"/>

<!--死信队列-->
<rabbit:queue name="dlx_queue"/>
<!--定向死信交换机-->
<rabbit:direct-exchange name="dlx_exchange" >
    <rabbit:bindings>
        <rabbit:binding key="dlx_ttl" queue="dlx_queue"></rabbit:binding>
        <rabbit:binding key="dlx_max" queue="dlx_queue"></rabbit:binding>
    </rabbit:bindings>
</rabbit:direct-exchange>

<!--测试超时的队列-->
<rabbit:queue name="test_ttl_queue">
    <rabbit:queue-arguments>
        <!--队列ttl为6秒-->
        <entry key="x-message-ttl" value-type="long" value="6000"/>
        <!--超时 消息 投递给 死信交换机-->
        <entry key="x-dead-letter-exchange" value="dlx_exchange"/>
    </rabbit:queue-arguments>
</rabbit:queue>

<!--测试超长度的队列-->
<rabbit:queue name="test_max_queue">
    <rabbit:queue-arguments>
        <!--队列ttl为6秒-->
        <entry key="x-max-length" value-type="long" value="2"/>
        <!--超时 消息 投递给 死信交换机-->
        <entry key="x-dead-letter-exchange" value="dlx_exchange"/>
    </rabbit:queue-arguments>
</rabbit:queue>

<!--定向测试交换机-->
<rabbit:direct-exchange name="my_exchange" >
    <rabbit:bindings>
        <rabbit:binding key="dlx_ttl" queue="test_ttl_queue">
        </rabbit:binding>
        <rabbit:binding key="dlx_max" queue="test_max_queue">
        </rabbit:binding>
    </rabbit:bindings>
</rabbit:direct-exchange>

发送消息测试

public class SenderDLX {
    public static void main(String[] args) {
        ClassPathXmlApplicationContext context = new
        ClassPathXmlApplicationContext("spring/spring-rabbitmq-producer-dlx.xml");
        RabbitTemplate rabbitTemplate = context.getBean(RabbitTemplate.class);
        // rabbitTemplate.convertAndSend("dlx_ttl", "测试超时".getBytes());
        rabbitTemplate.convertAndSend("dlx_max", "测试长度1".getBytes());
        rabbitTemplate.convertAndSend("dlx_max", "测试长度2".getBytes());
        rabbitTemplate.convertAndSend("dlx_max", "测试长度3".getBytes());
        System.out.println("消息已发出...");
        context.close();
    }
}

2.10 延迟队列

延迟队列:TTL + 死信队列的合体

死信队列只是一种特殊的队列,里面的消息仍然可以消费

spring-rabbitmq-consumer.xml

<!-- 监听死信队列 -->
<rabbit:listener-container connection-factory="connectionFactory" prefetch="3"
acknowledge="manual">
	<rabbit:listener ref="consumerListener" queue-names="dlx_queue" />
</rabbit:listener-container>

3 RabbitMQ集群

rabbitmq有3种模式,但集群模式是2种

单一模式:即单机情况不做集群,就单独运行一个rabbitmq而已(并不推荐使用)

普通模式:默认模式,以两个节点(A、B)为例来进行说明

当消息进入A节点的Queue后,consumer从B节点消费时,RabbitMQ会在A和B之间创建临时通道进行消息传输,把A中的消息实体取出并经过通过交给B发送给consumer

当A故障后,B就无法取到A节点中未消费的消息实体

如果做了消息持久化,那么得等A节点恢复,然后才可被消费
如果没有持久化的话,就会产生消息丢失的现象

镜像模式:非常经典的 mirror 镜像模式,保证 100% 数据不丢失。
高可靠性解决方案,主要就是实现数据的同步,一般来讲是 2 - 3 个节点实现数据同步
对于 100% 数据可靠性解决方案,一般是采用 3 个节点。
在实际工作中也是用得最多的,并且实现非常的简单,一般互联网大厂都会构建这种镜像集群模式

3.1 集群搭建

前置条件:准备两台linux,并安装好rabbitmq

集群步骤:

  1. 修改 /etc/hosts 映射文件

    1号服务器:

    127.0.0.1 A localhost localhost.localdomain localhost4
    localhost4.localdomain4
    ::1 A localhost localhost.localdomain localhost6
    localhost6.localdomain6
    192.168.204.141 A
    192.168.204.142 B
    

    2号服务器:

    127.0.0.1 B localhost localhost.localdomain localhost4
    localhost4.localdomain4
    ::1 B localhost localhost.localdomain localhost6
    localhost6.localdomain6
    192.168.204.141 A
    192.168.204.142 B
    
  2. 相互通信,cookie必须保持一致,同步 rabbitmq的cookie 文件:跨服务器拷贝 .erlang.cookie(隐藏文件,使用 ls -all 显示)

    [[email protected] opt]# scp /var/lib/rabbitmq/.erlang.cookie
    192.168.204.142:/var/lib/rabbitmq
    

    修改cookie文件,要重启服务器,reboot

  3. 停止防火墙,启动rabbitmq服务

    [[email protected] ~]# systemctl stop firewalld
    [[email protected] ~]# systemctl start rabbitmq-server
    
  4. 加入集群节点

    [[email protected] ~]# rabbitmqctl stop_app
    [[email protected] ~]# rabbitmqctl join_cluster [email protected]
    [[email protected] ~]# rabbitmqctl start_app
    
  5. 查看节点状态

    [[email protected] ~]# rabbitmqctl cluster_status
    
  6. 查看管理端

    搭建集群结构之后,之前创建的交换机、队列、用户都属于单一结构,在新的集群环境中是不能用的

    所以在新的集群中重新手动添加用户即可(任意节点添加,所有节点共享)

    [[email protected] ~]# rabbitmqctl add_user laosun 123123
    [[email protected] ~]# rabbitmqctl set_user_tags laosun administrator
    [[email protected] ~]# rabbitmqctl set_permissions -p "/" laosun ".*" ".*" ".*"
    

    注意:当节点脱离集群还原成单一结构后,交换机,队列和用户等数据 都会重新回来

3.2 镜像模式

将所有队列设置为镜像队列,即队列会被复制到各个节点,各个节点状态一致

语法:set_policy {name} {pattern} {definition}

name:策略名,可自定义

pattern:队列的匹配模式(正则表达式)

“^” 可以使用正则表达式,比如"^queue_" 表示对队列名称以“queue_”开头的所有队列进行镜像,而"^"表示匹配所有的队列

definition:镜像定义,包括三个部分ha-mode, ha-params, ha-sync-mode

ha-mode:(High Available,高可用)模式,指明镜像队列的模式,有效值为all/exactly/nodes,当前策略模式为 all,即复制到所有节点,包含新增节点

all:表示在集群中所有的节点上进行镜像
exactly:表示在指定个数的节点上进行镜像,节点的个数由ha-params指定
nodes:表示在指定的节点上进行镜像,节点名称通过ha-params指定

ha-params:ha-mode模式需要用到的参数
ha-sync-mode:进行队列中消息的同步方式,有效值为automatic和manual

[[email protected] ~]# rabbitmqctl set_policy xall "^" '{"ha-mode":"all"}'

3.3 HAProxy实现镜像队列的负载均衡

3.3.1 HAProxy

HA(High Available,高可用),Proxy(代理)
HAProxy是一款提供高可用性,负载均衡,并且基于TCP和HTTP应用的代理软件

HAProxy可以支持数以万计的并发连接
HAProxy可以简单又安全的整合进架构中,同时还保护web服务器不被暴露到网络上

3.3.2 HAProxy与Nginx

OSI:(Open System Interconnection:开放式系统互联 是把网络通信的工作分为7层,分别是物理层,数据链路层,网络层,传输层,会话层,表示层和应用层)

Nginx的优点

工作在OSI第7层,可以针对http应用做一些分流的策略
Nginx对网络的依赖非常小,理论上能ping通就就能进行负载功能,屹立至今的绝对优势
Nginx安装和配置比较简单,测试起来比较方便;
Nginx不仅仅是一款优秀的负载均衡器/反向代理软件,它同时也是功能强大的Web应用服务器

HAProxy的优点:

工作在网络4层和7层,支持TCP与Http协议,
它仅仅就只是一款负载均衡软件;单纯从效率上来讲HAProxy更会比Nginx有更出色的负载均
衡速度,在并发处理上也是优于Nginx的
支持8种负载均衡策略 ,支持心跳检测

性能上HA胜,功能性和便利性上Nginx胜

对于Http协议Haproxy处理效率比Nginx高。所以,没有特殊要求的时候或者一般场景,建议使用Haproxy来做Http协议负载

但如果是Web应用,那么建议使用Nginx

``

注意:当节点脱离集群还原成单一结构后,交换机,队列和用户等数据 都会重新回来

3.2 镜像模式

将所有队列设置为镜像队列,即队列会被复制到各个节点,各个节点状态一致

语法:set_policy {name} {pattern} {definition}

name:策略名,可自定义

pattern:队列的匹配模式(正则表达式)

“^” 可以使用正则表达式,比如"^queue_" 表示对队列名称以“queue_”开头的所有队列进行镜像,而"^"表示匹配所有的队列

definition:镜像定义,包括三个部分ha-mode, ha-params, ha-sync-mode

ha-mode:(High Available,高可用)模式,指明镜像队列的模式,有效值为all/exactly/nodes,当前策略模式为 all,即复制到所有节点,包含新增节点

all:表示在集群中所有的节点上进行镜像
exactly:表示在指定个数的节点上进行镜像,节点的个数由ha-params指定
nodes:表示在指定的节点上进行镜像,节点名称通过ha-params指定

ha-params:ha-mode模式需要用到的参数
ha-sync-mode:进行队列中消息的同步方式,有效值为automatic和manual

[[email protected] ~]# rabbitmqctl set_policy xall "^" '{"ha-mode":"all"}'

3.3 HAProxy实现镜像队列的负载均衡

3.3.1 HAProxy

HA(High Available,高可用),Proxy(代理)
HAProxy是一款提供高可用性,负载均衡,并且基于TCP和HTTP应用的代理软件

HAProxy可以支持数以万计的并发连接
HAProxy可以简单又安全的整合进架构中,同时还保护web服务器不被暴露到网络上

3.3.2 HAProxy与Nginx

OSI:(Open System Interconnection:开放式系统互联 是把网络通信的工作分为7层,分别是物理层,数据链路层,网络层,传输层,会话层,表示层和应用层)

Nginx的优点

工作在OSI第7层,可以针对http应用做一些分流的策略
Nginx对网络的依赖非常小,理论上能ping通就就能进行负载功能,屹立至今的绝对优势
Nginx安装和配置比较简单,测试起来比较方便;
Nginx不仅仅是一款优秀的负载均衡器/反向代理软件,它同时也是功能强大的Web应用服务器

HAProxy的优点:

工作在网络4层和7层,支持TCP与Http协议,
它仅仅就只是一款负载均衡软件;单纯从效率上来讲HAProxy更会比Nginx有更出色的负载均
衡速度,在并发处理上也是优于Nginx的
支持8种负载均衡策略 ,支持心跳检测

性能上HA胜,功能性和便利性上Nginx胜

对于Http协议Haproxy处理效率比Nginx高。所以,没有特殊要求的时候或者一般场景,建议使用Haproxy来做Http协议负载

但如果是Web应用,那么建议使用Nginx

相关标签: Java自学笔记