RabbitMq 学习笔记

浅浅的花香味﹌ 2022-02-20 12:32 213阅读 0赞

## RabbitMq ##

##### 队列模型 #####

*  simple queue
    
     *  一一对应，一个生产者对应一个消费者
 *  work queue
    
     *  一个生产者，多个消费者

##### work queue #####

###### 分发方式 ######

*  轮询分发
 *  公平分发

--------------------

#### 消息应答模式 ####

channel.basicConsume(QUEUE\_NAME, autoAck, consumer);

##### `autoAck = true` 自动确认模式 （默认值） #####

一旦rabbitmq将消息分发给消费者，就会从内存中删除，  
这种情况下，如果正在执行的消费者中断了，那正在处理的消息将丢失。

##### `autoAck = false` 手动模式 #####

如果有一个消费者挂掉，就会交付给其他消费者。rabbitmq支持消息应答，  
消费者发送一个消息告诉mq这个消息处理完成，则mq从内存中将其信息删除

##### `如果rabbitmq挂了，消息会丢失` #####

--------------------

##### 消息持久化 #####

channel.queueDeclare(QUEUE\_NAME,durable,false,false,null)

##### `durable=false` 不开启持久化 #####

rabbitmq 不允许重新定义（参数不同）一个已存在的队列

--------------------

### 交换机 ###

#### 订阅模式 `publish/subscribe` ####

**模型**

|||Q|||----[C1]
              /
    [P]----[X]
              \
               |||Q|||----[C2]

*  p : 生产者
 *  x ：交换机（转发器）
 *  q : 队列
 *  c : 消费者

##### 解读 #####

1.  一个生产者 多个消费者
2.  每个消费者都有自己的队列
3.  生产者没有直接把消息发送到队列 而是发送到 转发器/交换机/exchange
4.  每个队列都要绑定到交换机上
5.  生产者发送的消息经过交换机 到达队列就能实现一个消息被多个消费者消费

##### 生产者 #####

###### code ######

public class Send { 
        private static String EXCHANGE_NAME = "test_exchange_fanout";
        public static void main(String[] args) throws IOException, TimeoutException { 
            Connection connection = ConnectionUtils.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明交换机
            channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
    
            String msg = "hello ps";
    
            // 发布消息
            channel.basicPublish(EXCHANGE_NAME, "", null, msg.getBytes());
    
            System.out.println("Send:" + msg);
    
            channel.close();
            connection.close();
        }
    }

##### 无队列绑定至交换机时，发送消息后该消息会丢失，因为交换机没有存储能力 #####

##### 消费者 #####

###### code ######

`消费者1`

public class Recv1 { 
       private static String QUEUE_NAME = "test_queue_fanout_email";
       private static String EXCHANGE_NAME = "test_exchange_fanout";
       public static void main(String[] args) throws IOException, TimeoutException { 
           Connection connection = ConnectionUtils.getConnection();
           Channel channel = connection.createChannel();
    
           // 声明队列
           channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
           // 绑定交换机
           channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
    
           channel.basicQos(1);
    
           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, "utf-8");
                   System.out.println("consumer1-msg:" + msg);
    
                   try { 
                       Thread.sleep(1000);
                   } catch (InterruptedException e) { 
                       e.printStackTrace();
                   } finally { 
                       System.out.println(" [1] done ");
    
                       // 手动发送回执
                       channel.basicAck(envelope.getDeliveryTag(),false);
                   }
               }
           } ;
           // 关闭自动应答
           boolean autoAck = false;
           channel.basicConsume(QUEUE_NAME, autoAck, consumer);
       }
    }

`消费者2`

public class Recv2 { 
        private static String QUEUE_NAME = "test_queue_fanout_sms";
        private static String EXCHANGE_NAME = "test_exchange_fanout";
        public static void main(String[] args) throws IOException, TimeoutException { 
            Connection connection = ConnectionUtils.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 绑定交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
    
            channel.basicQos(1);
    
            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, "utf-8");
                    System.out.println("consumer2-msg:" + msg);
    
                    try { 
                        Thread.sleep(1000);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    } finally { 
                        System.out.println(" [2] done ");
    
                        // 手动发送回执
                        channel.basicAck(envelope.getDeliveryTag(),false);
                    }
                }
            } ;
            // 关闭自动应答
            boolean autoAck = false;
            channel.basicConsume(QUEUE_NAME, autoAck, consumer);
        }
    }

##### exchange 交换机 #####

接收生产者消息，向队列推送消息

`fanout` 不处理路由键

<msg>
    broker           |
    exchange       / | \
    bindings      /  |  \
                 #   #   #
    queue        #   #   #
                 #   #   #

`direct` 处理路由键

<msg>
    broker           |
    exchange           \
    bindings            \
                 #   #   #
    queue        #   #  key
                 #   #   #

--------------------

#### 路由模式 ####

x : type = direct
    
                 |||Q|||----[C1]
                /
             error
              /
    [P]----[X] -------info
            |   \       |
            |   error   |
            |     \     |
         warning-- -|||Q|||----[C2]

##### 生产者 #####

public class Send { 
        private static String EXCHANGE_NAME = "test_exchange_direct";
        public static void main(String[] args) throws IOException, TimeoutException { 
            Connection connection = ConnectionUtils.getConnection();
            Channel channel = connection.createChannel();
    
            // exchange
            channel.exchangeDeclare(EXCHANGE_NAME, "direct");
    
            String msg = "hello direct";
    
            // 发布消息
            String routingKey = "info";
            channel.basicPublish(EXCHANGE_NAME, routingKey, null, msg.getBytes());
    
            channel.close();
            connection.close();
        }
    }

##### 消费者1 #####

public class Recv1 { 
        private static String QUEUE_NAME = "test_queue_direct_1";
        private static String EXCHANGE_NAME = "test_exchange_direct";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            Connection connection = ConnectionUtils.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 绑定交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "error");
    
            channel.basicQos(1);
    
            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, "utf-8");
                    System.out.println("consumer1-msg:" + msg);
    
                    try { 
                        Thread.sleep(1000);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    } finally { 
                        System.out.println(" [1] done ");
    
                        // 手动发送回执
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    }
                }
            };
            // 关闭自动应答
            boolean autoAck = false;
            channel.basicConsume(QUEUE_NAME, autoAck, consumer);
        }
    }

##### 消费者2 #####

public class Recv2 { 
        private static String QUEUE_NAME = "test_queue_direct_2";
        private static String EXCHANGE_NAME = "test_exchange_direct";
    
        public static void main(String[] args) throws IOException, TimeoutException { 
            Connection connection = ConnectionUtils.getConnection();
            Channel channel = connection.createChannel();
    
            // 声明队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 绑定交换机
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "error");
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "info");
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "warning");
    
            channel.basicQos(1);
    
            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, "utf-8");
                    System.out.println("consumer2-msg:" + msg);
    
                    try { 
                        Thread.sleep(1000);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    } finally { 
                        System.out.println(" [2] done ");
    
                        // 手动发送回执
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    }
                }
            };
            // 关闭自动应答
            boolean autoAck = false;
            channel.basicConsume(QUEUE_NAME, autoAck, consumer);
        }
    }

--------------------

#### Topic exchange `将路由和某模式匹配` ####

message    usa.news  usa.weather  europe.news  europe.weather
    broker        |          |             |             |
                  | \      /   \         /   \         / |
    exchange      |   \  /       \     /       \     /   |
                  |   / \          \ /           \ /     |
                  | /     \        / \           / \     |
    queue     ||usa.*||    ||news||  ||*.weather||  ||europr.*||

模型图

*  \# 匹配多个
 *  \* 匹配一个

x : type = topic
    
                 |||Q|||----[C1]
                /
            *.orange
              /
    [P]----[X] -----*.*.rabbit
            |   \       |
            |  lazy.#   |
            |     \     |
          snack.*-- -|||Q|||----[C2]

--------------------

#### rabbitmq 消息确认机制（事务 + confirm） ####

在rabbitmq中，可通过持久化数据解决rabbitmq服务器异常造成的数据丢失问题

`问题:` 生产者将消息发送出去之后到底有没有到达rabbitmq服务器，默认情况是不知道的

`解决方式：`

*  AMQP 协议
 *  Confirm 模式

##### 事务机制 #####

`txSelect` `txCommit` `txRollback`

*  txSelect: 用户将当前channel设置成transation模式
 *  txCommit: 用于提交事务
 *  txRollback: 回滚事务

事务机制 由于过多的通信 降低了吞吐量

##### confirm模式 #####

同一个队列不能同时处于事务模式和confirm模式

`confirm模式实现原理`

`开启confirm模式` channel.confirmSelect()

code

*  同步

public class Send2 { 
        private static final String QUEUE_NAME = "test_queue_confirm";
        public static void main(String[] args) throws IOException, TimeoutException, InterruptedException { 
            Connection connection = ConnectionUtils.getConnection();
    
            Channel channel = connection.createChannel();
    
            // 定义队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 设为confirm模式
            channel.confirmSelect();
    
            String msg = "hello confirm ";
            // 批量发送
            for (int i = 0; i < 10; i++) { 
                channel.basicPublish("", QUEUE_NAME, null, (msg+i).getBytes());
            }
    
            if (!channel.waitForConfirms()) { 
                System.out.println("send failed");
            }else{ 
                System.out.println("send success");
            }
    
            channel.close();
            connection.close();
        }
    }

*  异步

public class Send3 { 
        private static final String QUEUE_NAME = "test_queue_confirm3";
        public static void main(String[] args) throws IOException, TimeoutException, InterruptedException { 
            Connection connection = ConnectionUtils.getConnection();
    
            Channel channel = connection.createChannel();
    
            // 定义队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            // 设为confirm模式
            channel.confirmSelect();
    
            // 未确认的消息标识
            final SortedSet<Long> confirmSet = Collections.synchronizedSortedSet(new TreeSet<Long>());
    
            channel.addConfirmListener(new ConfirmListener() { 
    
                // 没问题正常调ack
                @Override
                public void handleAck(long l, boolean b) throws IOException { 
                    if (b) { 
                        System.out.println("--handleAck--multiple");
                        confirmSet.headSet(l + 1).clear();
                    } else { 
                        System.out.println("--handleAck--multiple false");
                        confirmSet.remove(l);
                    }
                }
    
                // 出问题调nack
                @Override
                public void handleNack(long l, boolean b) throws IOException { 
                    if (b) { 
                        System.out.println("--handleNack--multiple");
                        confirmSet.headSet(l + 1).clear();
                    } else { 
                        System.out.println("--handleNack--multiple false");
                        confirmSet.remove(l);
                    }
                }
            });
    
            String msg = "hello confirm ";
    
            while (true) { 
                long seqNo = channel.getNextPublishSeqNo();
                channel.basicPublish("",QUEUE_NAME,null,msg.getBytes());
                confirmSet.add(seqNo);
            }
    
        }
    }