rabbitMq 基础知识

rabbitmq.png

基本概念

broker: 实体服务器
VirtualHost：缩小版的RabbitMq服务器，拥有自己独立的交换器、消息队列和相关的对象
Exchange:接受生产者的消息，并将这些消息路由到具体的Queue中
Binding:Exchange和Queue之间的关联
Queue:用来保存消息直到发送给消费者，它是消息的容器。
Channel:多路复用连接中的独立的双向数据流通道，因为建立和销毁TCP的Connection开销

rabbitmq对消息的保存方式

disk 后缀.rdp
a. 在发送时指定需要持久化或者服务器内存紧张时会将部分中的内存消息保存到磁盘中
b. 单个文件增加到16M后会生成新的文件
c. 文件中的消息被标记删除的比例达到阈值时会触发文件的合并，提高磁盘的利用率

Exchange 类型

订阅模式(Fanout Exchange)：

会将消息放到所有绑定到该exchange的队列上

rabbitMq_fanOut.png

public class ConnectionUtils {

    private static Connection connection;
    private static String lock = "aaa";

  /**
     * 获取rabbitmq连接
     * @return
     */
    public static Connection getConnection() {
        if (null != connection) {
            return connection;
        }
        synchronized (lock) {
            if (null != connection) {
                return connection;
            }

            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("host");
            connectionFactory.setUsername("userName");
            connectionFactory.setPassword("password");
            connectionFactory.setVirtualHost("/vhost");
            try {
                connection = connectionFactory.newConnection();
            } catch (IOException e) {
                throw new RuntimeException("IoException", e);
            } catch (TimeoutException e) {
                throw new RuntimeException("timeOutException", e);
            }
            return connection;
        }
    }

}

public class FanOutProducer {

    private static final String EXCHAGE_NAME = "test_exchange_li_fanout";

    public static void main(String[] args) throws IOException, TimeoutException {
        //建立连接
        Connection connection = ConnectionUtils.getConnection();
        Channel channel = connection.createChannel();

        //声明一个exchange,fanout类型，不持久化
        channel.exchangeDeclare(EXCHAGE_NAME, "fanout", false);

        //发送消息
        StringBuilder stringBuilder = new StringBuilder("message");
        for (int i = 0; i < 10; i++) {
            channel.basicPublish(EXCHAGE_NAME, "", null, stringBuilder.append(i).toString().getBytes("utf-8"));
        }

        //关闭连接
        channel.close();
        connection.close();
    }
}

public class FanOutConsumer01 {

    private static final String EXCHAGE_NAME = "test_exchange_li_fanout";
    private static final String QUEUE_NAME = "test_queue_Name_li_fanout";

    public static void main(String[] args) throws IOException {
        //建立链接
        Connection connection = ConnectionUtils.getConnection();
        Channel channel = connection.createChannel();

        //声明exchange和queue--exchange要和生产者一致
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        channel.exchangeDeclare(EXCHAGE_NAME, "fanout");

        //将queue绑定到exchange上
        channel.queueBind(QUEUE_NAME, EXCHAGE_NAME, "");

        //定义消费者
        Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
                throws IOException {
                String message = new String(body, "utf-8");
                System.out.println(message);
            }
        };

        //开始消费
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

public class FanOutConsumer02 {

    private static final String EXCHAGE_NAME = "test_exchange_li_fanout";
    private static final String QUEUE_NAME = "test_queue_Name_li_fanout02";
....
}

两个消费者都能打印如下
message0
message01
message012
message0123
message01234
message012345
message0123456
message01234567
message012345678
message0123456789

Direct Exchange

只会把消息routingkey一致的queue中

rabbitMq_binding.png

rabbitMq_Direct.png

public class DirectProducer {

    private static final String EXCHAGE_NAME = "test_exchange_li_direct";
    private static final String ROUTING_KEY = "direct_routing_key";

    public static void main(String[] args) throws IOException, TimeoutException {
        //建立连接
        Connection connection = ConnectionUtils.getConnection();
        Channel channel = connection.createChannel();

        //声明一个exchange,direct类型，不持久化
        channel.exchangeDeclare(EXCHAGE_NAME, "direct", false);

        //发送消息
        StringBuilder stringBuilder = new StringBuilder("message");
        for (int i = 0; i < 10; i++) {
            channel.basicPublish(EXCHAGE_NAME, ROUTING_KEY, null, stringBuilder.append(i).toString().getBytes("utf-8"));
        }

        //关闭连接
        channel.close();
        connection.close();
    }
}

public class DirectConsumer01 {

    private static final String EXCHAGE_NAME = "test_exchange_li_direct";
    private static final String QUEUE_NAME = "test_queue_Name_li_direct";
    private static final String ROUTING_KEY = "direct_routing_key";

    public static void main(String[] args) throws IOException {
        //建立链接
        Connection connection = ConnectionUtils.getConnection();
        Channel channel = connection.createChannel();

        //声明exchange和queue--exchange要和生产者一致
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        channel.exchangeDeclare(EXCHAGE_NAME, "direct");

        //将queue绑定到exchange上
        channel.queueBind(QUEUE_NAME, EXCHAGE_NAME, ROUTING_KEY);

        //定义消费者
        Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
                throws IOException {
                String message = new String(body, "utf-8");
                System.out.println(message);
            }
        };

        //开始消费
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

public class DirectConsumer02 {

    private static final String EXCHAGE_NAME = "test_exchange_li_direct";
    private static final String QUEUE_NAME = "test_queue_Name_li_direct01";
    private static final String ROUTING_KEY = "direct_routing_key02";
....
}

运行完成，只有DirectConsumer01能收到消息
message0
message01
message012
message0123
message01234
message012345
message0123456
message01234567
message012345678
message0123456789

Topic Exchange

对key进行模式匹配后进行投递，符号”#”匹配一个或多个词，符号””匹配正好一个词。例如”abc.#”匹配”abc.def.ghi”，”abc.”只匹配”abc.def”

rabbitMq_Topic.png

public class TopicProducer {

    private static final String EXCHAGE_NAME = "test_exchange_li_topic";

    private static final String ROUTING_KEY = "test.routingkey.01";

    public static void main(String[] args) throws IOException, TimeoutException {
        //建立连接
        Connection connection = ConnectionUtils.getConnection();
        Channel channel = connection.createChannel();

        //声明一个exchange,topic类型，不持久化
        channel.exchangeDeclare(EXCHAGE_NAME, "topic", false);

        //发送消息
        StringBuilder stringBuilder = new StringBuilder("message");
        for (int i = 0; i < 10; i++) {
            channel.basicPublish(EXCHAGE_NAME,ROUTING_KEY, null, stringBuilder.append(i).toString().getBytes("utf-8"));
        }
        //关闭连接
        channel.close();
        connection.close();
    }
}

public class TopicConsumer01 {

    private static final String EXCHAGE_NAME = "test_exchange_li_topic";
    private static final String QUEUE_NAME = "test_queue_Name_li_topic_01";
    private static final String ROUTING_KEY = "test.routingkey.#";

    public static void main(String[] args) throws IOException {
        //建立链接
        Connection connection = ConnectionUtils.getConnection();
        Channel channel = connection.createChannel();

        //声明exchange和queue--exchange要和生产者一致
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        channel.exchangeDeclare(EXCHAGE_NAME, "topic");

        //将queue绑定到exchange上
        channel.queueBind(QUEUE_NAME, EXCHAGE_NAME, ROUTING_KEY);

        //定义消费者
        Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
                throws IOException {
                String message = new String(body, "utf-8");
                System.out.println(message);
            }
        };

        //开始消费
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

为了节省篇幅，
public class TopicConsumer02 {

    private static final String EXCHAGE_NAME = "test_exchange_li_topic";
    private static final String QUEUE_NAME = "test_queue_Name_li_topic_02";
    private static final String ROUTING_KEY = "test.routingkey.*";
......
}

public class TopicConsumer03 {

    private static final String EXCHAGE_NAME = "test_exchange_li_topic";
    private static final String QUEUE_NAME = "test_queue_Name_li_topic_03";
    private static final String ROUTING_KEY = "test.routingkey";
....
}

输出结果只有TopicConsumer01和TopicConsumer02有日志输出
message0
message01
message012
message0123
message01234
message012345
message0123456
message01234567
message012345678
message0123456789

内容的持久化`durable 表示持久化`

Queue的持久化

com.rabbitmq.client.Channel

Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,
                                 Map<String, Object> arguments) throws IOException;

Message的持久化

void basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body) throws IOException;


public static class BasicProperties extends com.rabbitmq.client.impl.AMQBasicProperties {
        private String contentType;
        private String contentEncoding;
        private Map<String,Object> headers;
        private Integer deliveryMode;
        private Integer priority;
        private String correlationId;
        private String replyTo;
        private String expiration;
        private String messageId;
        private Date timestamp;
        private String type;
        private String userId;
        private String appId;
        private String clusterId;

其中deliveryMode=2表示持久化

Exchange的持久化

com.rabbitmq.client.Channel

Exchange.DeclareOk exchangeDeclare(String exchange,
                                              String type,
                                              boolean durable,
                                              boolean autoDelete,
                                              boolean internal,
                                              Map<String, Object> arguments) throws IOException;

就算上面三个的持久化方式都开启了，也不能保证消息在使用过程中完全不丢失，如果消费者autoAck=true,在收到消息之后自动确认了，但是处理时服务崩了，就会导致消息的丢失,所以需要确认机制的支持

消息投递的确认模式

默认情况下，生产者投递消息后，broker时不会做出任何返回的
解决方式如下：
1. 使用Amqp协议中的事务机制效率低，影响吞吐量
2. 将信道channel设置成确认模式

channel信道的确认模式

channel设置成确认模式之后，所有提交的消息都会被分配一条唯一的ID，当消息被投递到匹配的队列中，信道会向生产者发出确认消息，并且消息中带上这个Id。确认模式是异步的，生产者可以发送完一条消息后继续发送下一条消息。调用channel的confirmSelect方法开启确认模式

普通方式，发送完成之后调用waitForConfirms
异步回调模式,addConfirmListener注册回调函数

消息消费的应答模式

autoAck，如果等于true，会在消息发送过来之后自动响应--队列会将该消息删除，可能会导致消息消费失败了，但是消息已经被删除的情况
autoAck=false,需要业务逻辑在处理完成之后，调用channel.basicAck做显示的响应

消费者获取消息时，可以指定预取的消息数量
通过channel的basicQos方法设置

rabbitmq 的死信队列`相当于为一个队列设置一个备用的队列，在出现以下情况的时候将所谓的死亡信息推送到死亡信息队列`

消息被拒绝（basic.reject/ basic.nack）并且不再重新投递 requeue=false
消息超期 (rabbitmq Time-To-Live -> messageProperties.setExpiration())
队列超载
具体内容参考：https://my.oschina.net/u/2948566/blog/1626278

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 204,590评论 6赞 478
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 86,808评论 2赞 381
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 151,151评论 0赞 337
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 54,779评论 1赞 277
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 63,773评论 5赞 367
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 48,656评论 1赞 281
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 38,022评论 3赞 398
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 36,678评论 0赞 258
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 41,038评论 1赞 299
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 35,659评论 2赞 321
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 37,756评论 1赞 330
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 33,411评论 4赞 321
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 39,005评论 3赞 307
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 29,973评论 0赞 19
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 31,203评论 1赞 260
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 45,053评论 2赞 350
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 42,495评论 2赞 343