rabbitmq五种消息模型和消息持久化

一、work消息模型-工作队列模型

1.工作队列的含义
主要思想就是避免执行资源密集型任务时，必须等待它执行完成。相反我们稍后完成任务，我们将任务封装为消息并将其发送到队列。 在后台运行的工作进程将获取任务并最终执行作业。当你运行许多消费者时，任务将在他们之间共享，但是一个消息只能被一个消费者获取。两个消费者消费数量相同。

2.弊端问题

• 消费者1比消费者2的效率要低，一次任务的耗时较长
• 然而两人最终消费的消息数量是一样的
• 消费者2大量时间处于空闲状态，消费者1一直忙碌3.解决弊端 让能者多劳channel.basicQos(5);设置客户端最多接收未被ack的消息的个数4.代码展示 1.生产者

publicclassSend{privatestaticfinalStringQUEUE_NAME="work_queue";publicstaticvoidmain(String[] args)throwsIOException,TimeoutException,InterruptedException{Connection con =ConnectionUtil.getCon();Channel channel = con.createChannel();
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);for(int i=0;i<50;i++){String message ="task...."+i;
            channel.basicPublish("",QUEUE_NAME,null,message.getBytes());System.out.println("send : "+message);Thread.sleep(i *2);}

        channel.close();
        con.close();}}

2.消费者1

publicclassRecv1{privatestaticfinalStringQUEUE_NAME="work_queue";publicstaticvoidmain(String[] args)throwsIOException,TimeoutException{Connection con =ConnectionUtil.getCon();Channel channel = con.createChannel();
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        channel.basicQos(5);DefaultConsumer consumer =newDefaultConsumer(channel){@OverridepublicvoidhandleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throwsIOException{String msg =newString(body);System.out.println("消费者1 received: "+msg);try{//模拟消费耗时Thread.sleep(1000);}catch(InterruptedException e){
                    e.printStackTrace();}
                channel.basicAck(envelope.getDeliveryTag(),false);}};

        channel.basicConsume(QUEUE_NAME,false,consumer);}}

3.消费者2

publicclassRecv2{privatestaticfinalStringQUEUE_NAME="work_queue";publicstaticvoidmain(String[] args)throwsIOException,TimeoutException{Connection con =ConnectionUtil.getCon();Channel channel = con.createChannel();
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);DefaultConsumer consumer =newDefaultConsumer(channel){@OverridepublicvoidhandleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throwsIOException{String msg =newString(body);System.out.println("消费者2 received: "+msg);

                channel.basicAck(envelope.getDeliveryTag(),false);}};

        channel.basicConsume(QUEUE_NAME,false,consumer);}}

二、订阅模型分类

1.订阅模型的含义
在这一部分，我们将做一些完全不同的事情 - 我们将会传递一个信息给多个消费者
特点：
​ 1、1个生产者，多个消费者
​ 2、每一个消费者都有自己的一个队列
​ 3、生产者没有将消息直接发送到队列，而是发送到了交换机
​ 4、每个队列都要绑定到交换机
​ 5、生产者发送的消息，经过交换机到达队列，实现一个消息被多个消费者获取的目的

2.交换机的作用和类型
交换机一方面：接收生产者发送的消息。另一方面：知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。

• Fanout：广播，将消息交给所有绑定到交换机的队列
• Direct：定向，把消息交给符合指定routing key 的队列
• Topic：通配符，把消息交给符合routing pattern（路由模式） 的队列

三、订阅模型-Fanout

1.广播模式的流程

• 1） 可以有多个消费者
• 2） 每个消费者有自己的queue（队列）
• 3） 每个队列都要绑定到Exchange（交换机）
• 4） 生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定。
• 5） 交换机把消息发送给绑定过的所有队列
• 6） 队列的消费者都能拿到消息。实现一条消息被多个消费者消费2.代码展示 1.生产者

publicclassSend{privatefinalstaticStringEXCHANGE_NAME="fanout_exchange";publicstaticvoidmain(String[] args)throwsIOException,TimeoutException{Connection con =ConnectionUtil.getCon();Channel channel = con.createChannel();// 声明exchange，指定类型为fanout
        channel.exchangeDeclare(EXCHANGE_NAME,"fanout");String message ="hello everyone";// 发布消息到Exchange
        channel.basicPublish(EXCHANGE_NAME,"",null,message.getBytes());System.out.println(" Sent : "+ message +"'");

        channel.close();
        con.close();}}

2.消费者1

publicclassRecv1{privatefinalstaticStringQUEUE_NAME="fanout_queue_1";privatefinalstaticStringEXCHANGE_NAME="fanout_exchange";publicstaticvoidmain(String[] args)throwsIOException,TimeoutException{Connection con =ConnectionUtil.getCon();Channel channel = con.createChannel();
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);// 绑定队列到交换机
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"");DefaultConsumer consumer =newDefaultConsumer(channel){@OverridepublicvoidhandleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throwsIOException{String msg =newString(body);System.out.println("消费者1 ："+msg);}};// 监听队列，自动返回完成
        channel.basicConsume(QUEUE_NAME,true,consumer);}}

2.消费者2

publicclassRecv2{privatefinalstaticStringQUEUE_NAME="fanout_queue_2";privatefinalstaticStringEXCHANGE_NAME="fanout_exchange";publicstaticvoidmain(String[] args)throwsIOException,TimeoutException{Connection con =ConnectionUtil.getCon();Channel channel = con.createChannel();
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"");DefaultConsumer consumer =newDefaultConsumer(channel){@OverridepublicvoidhandleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throwsIOException{String msg =newString(body);System.out.println("消费者2 ："+msg);}};
        channel.basicConsume(QUEUE_NAME,true,consumer);}}

四、订阅模型-Direct

1.定向订阅模型含义
在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。
在Direct模型下，队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）
消息的发送方在向Exchange发送消息时，也必须指定消息的routing key。

• 生产者，向Exchange发送消息，发送消息时，会指定一个routing key。 -Exchange（交换机），接收生产者的消息，然后把消息递交给 与routing key完全匹配的队列
• 消费者，其所在队列指定了需要routing key 为 error 的消息
• 消费者，其所在队列指定了需要routing key 为 info、error、warning 的消息2.代码展示 1.生产者

publicclassSend{privatefinalstaticStringEXCHANGE_NAME="direct_exchange";publicstaticvoidmain(String[] args)throwsIOException,TimeoutException{Connection con =ConnectionUtil.getCon();Channel channel = con.createChannel();
        channel.exchangeDeclare(EXCHANGE_NAME,"direct");String message ="插入商品";
        channel.basicPublish(EXCHANGE_NAME,"insert",null,message.getBytes());System.out.println("send : "+message);

        channel.close();
        con.close();}}

2.消费者1

publicclassRecv1{privatefinalstaticStringQUEUE_NAME="direct_queue_1";privatefinalstaticStringEXCHANGE_NAME="direct_exchange";publicstaticvoidmain(String[] args)throwsIOException,TimeoutException{Connection con =ConnectionUtil.getCon();Channel channel = con.createChannel();
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"update");
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"delete");DefaultConsumer consumer =newDefaultConsumer(channel){@OverridepublicvoidhandleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throwsIOException{String msg =newString(body);System.out.println("消费者1接收: "+msg);}};

        channel.basicConsume(QUEUE_NAME,true,consumer);}}

2.消费者2

publicclassRecv2{privatefinalstaticStringQUEUE_NAME="direct_queue_2";privatefinalstaticStringEXCHANGE_NAME="direct_exchange";publicstaticvoidmain(String[] args)throwsIOException,TimeoutException{Connection con =ConnectionUtil.getCon();Channel channel = con.createChannel();
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"insert");
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"update");
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"delete");DefaultConsumer consumer =newDefaultConsumer(channel){@OverridepublicvoidhandleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throwsIOException{String msg =newString(body);System.out.println("消费者2接收: "+msg);}};

        channel.basicConsume(QUEUE_NAME,true,consumer);}}

五、订阅模型-Topic

1.通配符订阅含义
Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符！
2…通配符规则

• #：匹配一个或多个词
• *：匹配不多不少恰好1个词
• audit.#：能够匹配audit.irs.corporate 或者 audit.irs
• audit.*：只能匹配audit.irs

3.代码展示
1.生产者

publicclassSend{privatefinalstaticStringEXCHANGE_NAME="topic_exchange";publicstaticvoidmain(String[] args)throwsIOException,TimeoutException{Connection con =ConnectionUtil.getCon();Channel channel = con.createChannel();
        channel.exchangeDeclare(EXCHANGE_NAME,"topic");String message ="商品新增";
        channel.basicPublish(EXCHANGE_NAME,"item.insert",null,message.getBytes());System.out.println("send : "+message);
        channel.close();
        con.close();}}

2.消费者1

publicclassRecv1{privatefinalstaticStringQUEUE_NAME="topic_queue_1";privatefinalstaticStringEXCHANGE_NAME="topic_exchange";publicstaticvoidmain(String[] args)throwsIOException,TimeoutException{Connection con =ConnectionUtil.getCon();Channel channel = con.createChannel();
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"item.insert");
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"item.delete");DefaultConsumer consumer =newDefaultConsumer(channel){@OverridepublicvoidhandleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throwsIOException{String msg =newString(body);System.out.println("消费者1接收: "+msg);}};

        channel.basicConsume(QUEUE_NAME,true,consumer);}}

3.消费者2

publicclassRecv2{privatefinalstaticStringQUEUE_NAME="topic_queue_2";privatefinalstaticStringEXCHANGE_NAME="topic_exchange";publicstaticvoidmain(String[] args)throwsIOException,TimeoutException{Connection con =ConnectionUtil.getCon();Channel channel = con.createChannel();
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"item.*");DefaultConsumer consumer =newDefaultConsumer(channel){@OverridepublicvoidhandleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body)throwsIOException{String msg =newString(body);System.out.println("消费者2接收: "+msg);}};

        channel.basicConsume(QUEUE_NAME,true,consumer);}}

六、持久化

1.持久化含义

1. 消费者的ACK机制。可以防止消费者丢失消息。
2. 但是，如果在消费者消费之前，MQ就宕机了，消息就没了。
3. 要将消息持久化，前提是：队列、Exchange都持久化

2.代码展示
1.交换机持久化

channel.exchangeDeclare(EXCHANGE_NAME,"topic",true);

2.队列持久化

channel.queueDeclare(QUEUE_NAME,true,false,false,null);

3.消息持久化

channel.basicPublish(EXCHANGE_NAME,"item.insert",MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());

上面是五种消息模型，以及持久化，rabbitmq到此结束不足之处，望海涵