【MQ】RubbitMq_rubbitmp

背景

rubbitmq是有erlang语言开发的基于AMQP的开源的消息队列

场景

（1）异步处理

（2）应用解耦

（3）流量削峰

特点

（1）持久化机制
exchange，queue : durable = true
deliveryMode=2 持久化 /1 非持久化

（2）事务机制
txSelect()选择事务 txCommit()提交事务 txRollback()回滚事务

channel.txSelect();
channel.basicPublish(ConfirmConfig.exchangeName, ConfirmConfig.routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, ConfirmConfig.msg_10B.getBytes());
channel.txCommit();
1
2
3

事务的确认机制：

client发送Tx.Select
broker发送Tx.Select-Ok(之后publish)
client发送Tx.Commit
broker发送Tx.Commit-Ok
1
2
3
4

事务的消耗很大，所以一般使用消息确认机制

（3）消息确认机制
消息队列向消费者推送消息，如果消息得到处理，需要消费者向队列发一个确认，然后队列可以删除这个消息；如果没有返回确认，mq没有使用超时机制，只通过consumer的中断来判断消息是否得到处理，如果一个consumer退出时还没有返回确认，队列会重新把消息发给另一个consumer。保证不会丢失数据

（4）公平分发
channel.basic_qos(prefetch_count) = 1

AMQP组件

AMQP是一个抽象的消息通讯协议

• server(broker):mq
• virtural host: 权限控制
• exchange：交换机，根据binding规则将消息路由到指定的queue上。分为三类：direct exchange， funout exchange， topic exchange
• message queue: 消息队列
• message：header+ body 数据主要在body里
• binding:连接exchange 和 message queue。会维护一张路由表 binding key routing key
• connection:生产者，消费者和broker之间的tcp连接
• channel:信道 因为connection是tcp连接，而tcp连接昂贵，一个connection可以连接多个channel
• command:命令 basePublish baseConsume

6种实现

（1）最简单的mq

生成者：

public class Send {
    private final static String QUEUE_NAME="hello";

    public static void main(String[] args) throws Exception{
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        String message = "hello world!";
        channel.basicPublish("",QUEUE_NAME,null,message.getBytes("UTF-8"));
        System.out.println("[x] Send '" + message + "'");

        channel.close();
        connection.close();


    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

消费者：

public class Recv {
   
    private final static String QUEUE_NAME = "hello";

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        System.out.println("[*] Waiting for messages. To exit press CTRL + C");

        Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throws IOException{
                String message = new String(body,"UTF-8");
                System.out.println("[x] Received '" + message +"'");
            }
        };
        //true代表自动消息确认开启，如果消费者没有确认发给别人，如果确认了队列删除数据
        channel.basicConsume(QUEUE_NAME,true,consumer);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

channel.basicConsume(队列，true，消费者)； true 表示开启消息确认机制
（2）工作队列–简单消息分发

生产者：

public class NewTask {
    private static final String TASK_QUEUE_NAME = "task_queue";

    public static void main(String[] args) throws Exception {
        String[] para = new String[4];
        para[0]="hello...";
        para[1]="Daniel...";
        para[2]="Hi..";
        para[3]="Summer..";
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        //true表示队列数据可持久化
        channel.queueDeclare(TASK_QUEUE_NAME,true,false,false,null);

        String message = getMessage(para);

        //持久化
        channel.basicPublish("",TASK_QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes("UTF-8"));
        System.out.println("[x] Sent '" + message + "'");

        channel.close();
        connection.close();
    }

    private static String getMessage(String[] strings){
        if(strings.length < 1)
            return "Hello world";
        return joinStrings(strings," ");
    }

    private static String joinStrings(String[] strings,String delimiter){
        int length = strings.length;
        if(length == 0 ) return "";
        StringBuilder stringBuilder = new StringBuilder(strings[0]);
        for(int i = 1; i < length; i++){
            stringBuilder.append(delimiter).append(strings[i]);
        }
        return stringBuilder.toString();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

• 持久化两个参数：
  durable deliveryMode
• 队列声明：
  channel.queueDeclare(TASK_QUEUE_NAME,true,false,false,null)

@ 第一个参数：队列名
@第二个参数：durable 持久化

• 发送给队列：
  channel.basicPublish(“”,TASK_QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes(“UTF-8”));

@第一个参数：交换机名
@第二个参数：队列名
@第三个参数：deliveryMode(1:不持久化；2：持久化) MessageProperties.PERSISTENT_TEXT_PLAIN：可持久化
@第四个参数：message

消费者：

public class Worker {
   
    private static final String TASK_QUEUE_NAME="task_queue";

    public static void main(String[] args) throws Exception{
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("localhost");
        final Connection connection=connectionFactory.newConnection();
        final Channel channel=connection.createChannel();

        channel.queueDeclare(TASK_QUEUE_NAME,true,false,false,null);
        System.out.println("[*] Waiting for messages. To exit press CTRL+C");

        //公平派遣，负载均衡
        channel.basicQos(1);

        final Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag,Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException{
                String message = new String(body,"UTF-8");

                System.out.println("[x] Received '" + message +"'");
                try{
                    doWork(message);
                }finally{
                    System.out.println("[x] Done");
                    //手动进行消息确认
                    channel.basicAck(envelope.getDeliveryTag(),false);
                }
            }
        };
        channel.basicConsume(TASK_QUEUE_NAME,false,consumer);
    }

    private static void doWork(String task){
        for(char ch : task.toCharArray()){
            
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37