了解一下RabbitMQ – palapala

RabbitMQ概述

RabbitMQ是遵从AMQP协议的 通信协议都设计到报文交互,换句话说RabbitMQ就是AMQP协议的Erlang的实现。

AMQP说到底还是一个通信协议从low-level层面举例来说,AMQP本身是应用层的协议,其填充于TCP协议的数据部分。

从high-level层面来说,AMQP是通过协议命令进行交互的。命令类似HTTP中的方法(GET PUT POST DELETE等)。

信道(Channel)在AMQP是一个很重要的概念,大多数操作都是在信道这个层面展开的

我们完全可以用Connection就能完成信道的工作,为什么还要引入信道?

试想:一个程序中有很多个线程需要从RabbitMQ中消费消息,或者生产消息,那么必然需要建立很多个Connection,也就是多个TCP连接。

建立和销毁TCP连接开销很昂贵。所以RabbitMQ采用类似NIO的做法,选择TCP连接复用不仅可以减少性能开销,同时也便于管理。

发布订阅模式

广播模式 topic

  所谓广播指的是一条消息将被所有的消费者进行处理。

直连模式 director

  直连模式的特点主要就是routingkey的使用,如果现在该消息就要求指定一个具备有指定Routingkey的操作者进行处理,那么只需要两个的Routingkey匹配即可。

  可以将Routingkey比喻一个唯一标记,这样就可以将消息准确的推送到消费者手中了。

主题模式 fanout

  主题模式类似于广播模式与直连模式的整合操作,所有的消费者都可以接收到主题信息,但是如果要想进行正确的处理,则一定需要有一个合适的Routingkey完成操作。

交换器相当于投递包裹的邮箱,Routingkey相当于包裹的地址,BindingKey相当于包裹的目的地。

当填写在包裹上的地址和要投递的地址相匹配时,那么这个包裹就会正确投递到目的地,最后这个目的地的主人(队列)可以保留这个包裹。

如果填写地址出错,邮递员不能正确的投递到目的地,包裹可能被退回给寄件人,也有可能被丢弃。

RabbitMQ官方文档和API都把Routingkey和BingdingKey都看做Routingkey下面代码中红色部分 就都当Routingkey使用

消息生产者

public class MessageProducer {
    private static final String EXCHANGE_NAME ="com.sunkun.topic";//消息队列名称
    private static final String HOST="192.168.1.105";
    private static final int PORT=5672;
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();//建立一个连接工厂
        factory.setHost(HOST);
        factory.setPort(PORT);
        factory.setUsername("sunkun");
        factory.setPassword("123456");
        //factory.setVirtualHost(virtualHost) 使用虚拟主机的最大好处 可以区分不同用户的操作空间  每一个虚拟主机有一个自己的空间管理
        Connection conn = factory.newConnection();//定义一个新的RabbitMQ的连接
        Channel channel = conn.createChannel();//创建一个通讯的通道
        //定义该通道要使用的队列名称 此时队列已经创建过了
        //第一个参数 队列名称(这个队列可能存在也可能不存在)
        //第二个参数 是否持久保存
        //第三个参数 此队列是否为专用的队列信息
        //第四个参数 是否允许自动删除
        //channel.queueDeclare(QUENE_NAME, true, false, true,null);
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");
        long start = System.currentTimeMillis();
        System.out.println("消息开始"+start);
        for(int i=0;i<1000;i++){
            String message = "sk - "+i;
            if(i%2==0){
                //MessageProperties.PERSISTENT_TEXT_PLAIN 消息持久化
                channel.basicPublish(EXCHANGE_NAME, "sk1", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());//进行消息发送
            }else{
                channel.basicPublish(EXCHANGE_NAME, "sk2", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());//进行消息发送
            }

        }
        long end = System.currentTimeMillis();
        System.out.println("消息花费时间"+(end-start));
        channel.close();
    }
}

消息消费者

public class MessageConsumer {
    private static final String EXCHANGE_NAME ="com.sunkun.topic";//消息队列名称
    private static final String HOST="192.168.1.105";
    private static final int PORT=15672;
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();//建立一个连接工厂
        factory.setHost(HOST);
        factory.setPort(PORT);
        factory.setUsername("sunkun");
        factory.setPassword("123456");
        Connection conn = factory.newConnection();//定义一个新的RabbitMQ的连接
        Channel channel = conn.createChannel();//创建一个通讯的通道
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");
        String queueName = channel.queueDeclare().getQueue();//通过通道获取一个队列名称
        channel.queueBind(queueName, EXCHANGE_NAME, "sk2");//进行绑定处理
        //在RabbitMQ里面,所有的消费者信息是通过一个回调方法完成的
        Consumer consumer = new DefaultConsumer(channel){//需要复写指定的方法实现消息处理
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                    BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body);
                System.out.println("消费者sk2:"+message);//可以启动多个消费者
                super.handleDelivery(consumerTag, envelope, properties, body);
            }
        };
        channel.basicConsume(queueName,consumer);
    }
}

RabbitMQ如何保证消息的可靠性

1)持久化

持久化可以提高RabbitMQ的可靠性,防止在异常情况(重启,关闭,宕机)下的数据丢失。

持久化可分为三个部分:交换器的持久化,队列的持久化和消息的持久化。

交换器的持久化:是通过声明交换器时将druable参数设置为true来实现的。如果交换器不设置持久化,那么在RabbitMQ重启之后相关的交换器元数据会丢失,不过消息不会丢失,只是不能将消息发送到这个交换器中了。

对于一个长期使用的交换器来说,建议其置为持久化。(消息不直接往队列发,往exchange发送 可以实现广播模式)

队列的持久化:是通过声明队列时将durable参数置为true实现的。如果队列不设置持久化,那么在RabbitMQ服务重启之后,相关队列的元数据会丢失,此时数据也会丢失。

消息的持久化:因为队列的持久化能保证其本身的元数据不会因为异常情况而丢失,但是不能保证内部存储的消息不会丢失。要确保消息不会丢失,需求将其设置为持久化。

通过将消息的投递模式(MessageProperties.PERSISTENT_TEXT_PLAIN)即可实现消息的持久化

2)集群

RabbitMQ的集群本身不带有所谓的HA机制以及负载均衡机制

本文主要讲镜像队列

在持久化的消息正确存入到RabbitMQ之后 还需要一段时间(虽然时间很短,但不可忽视)才能存入磁盘中,如果这段时间发生了宕机,消息保存还没来得及落盘,那么这些消息将会丢失。

这里可以引入RabbitMQ的镜像队列机制,相当于配置了副本,如果主节点(master)在此特殊时间内挂掉,可以自动切换到从节点(Slave),在实际生产环境中的关键业务队列都会设置镜像队列。

提醒:所谓的镜像队列只是进行数据的副本而已,在所谓的RabbitMQ集群里面并不支持HA机制以及所谓的负载均衡,如果说现在一台主机挂掉了,那么其他主机肯定无法进行合理读取的。

如果想要安全的使用RabbitMQ就要继续追加负载均衡组件,列如HAProxy LVS等等,如果要保证负载均衡组件的高可用,还应该继续追加KeepAlive组件(就像tomcat实现负载均衡 需要nginx一样)。

3)生产者确认

除上面两个问题外 我们还遇到一个新问题:当消息的生产者将消息发送出去之后,消息到底有没有正确的到达服务器呢?

如果消息到达服务器之前就丢失,那么持久化也解决不了问题,因为消息就没有到达服务器,何谈持久化呢。

通常会有两种方法解决此问题一时事物机制,只有消息被成功接收,事物才能提交成功,否则便可在捕获异常之后进行事物回滚,于此同时可以进行消息重发。

但使用事物机制会大大降低RabbitMQ的性能,我们一般采取发送方确认机制。

发送方确认机制:生产者将信道设置成confirm模式,一旦信道进入confirm模式,所有在该信道上面发布的消息都会被指派一个唯一的ID(从1开始),

一旦消息被投递到所有的匹配队列之后,RabbitMQ就会发送一个确认给生产者(包含消息的唯一ID),这就使得生产者知晓消息已经到达目的地了。

如果消息和队列是持久化的,那么消息确认会在消息写入磁盘后发出。



You must enable javascript to see captcha here!

Copyright © All Rights Reserved · Green Hope Theme by Sivan & schiy · Proudly powered by WordPress

无觅相关文章插件,快速提升流量