SpringBoot整合Rabbitmq之===RabbitMQ的 Work 模型（无废话版）--------------------------------------------（三）

红太狼 2022-12-27 11:20 57阅读 0赞

# 代码地址：[https://github.com/heidaodageshiwo/ZqRabbitMq][https_github.com_heidaodageshiwo_ZqRabbitMq] #

#  ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1bmhlaWRhb2Rl_size_16_color_FFFFFF_t_70][] #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1bmhlaWRhb2Rl_size_16_color_FFFFFF_t_70 1][]

生产者1个

package org.example.Work;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.MessageProperties;
    import java.io.IOException;
    import org.example.RabbitMQUtils;
    import org.junit.Test;
    
    /**
     * java类简单作用描述
     *
     * @ProjectName: rabbitmq
     * @Package: org.example.Work
     * @ClassName: Provider1
     * @Description: java类作用描述
     * @Author: zhangq
     * @CreateDate: 2020-12-10 18:06
     * @UpdateUser: zhangq
     * @UpdateDate: 2020-12-10 18:06
     * @UpdateRemark: The modified content
     * @Version: 1.0 *
     */
    public class Provider1 {
      @Test
      public void sendmessage() throws IOException {
        //获取连接对象
        Connection connection= RabbitMQUtils.getConnection();
        //获取连接中通道
        Channel channel = connection.createChannel();
        //通道绑定对应消息队列
        //参数1 队列名称如果队列不存在则自动创建
        //参数2 用来定义队列特性是否要持久化 true持久化
        //参数3 是否是独占队列 true 独占
        //参数4 消费者消费完消息是否自动删除队列 true自动删除
        //参数5 额外参数
        channel.queueDeclare("work",true,false,false,null);
        //发布消息
        //参数1 交换机名称 参数2队列名称 参数3参数 参数4 消息的具体内容
    //    channel.basicPublish("","hello",null,"张强测试".getBytes());
        for (int i = 0; i < 10; i++) {
          channel.basicPublish("","work", MessageProperties.PERSISTENT_TEXT_PLAIN,("张强测试work:"+i).getBytes());
        }
    
    
        RabbitMQUtils.closeConnectionFactory(channel, connection);
    
    
      }
    
    }

消费者1：

package org.example.Work;
    
    import com.rabbitmq.client.AMQP.BasicProperties;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.DefaultConsumer;
    import com.rabbitmq.client.Envelope;
    import java.io.IOException;
    import org.example.RabbitMQUtils;
    
    /**
     * java类简单作用描述
     *
     * @ProjectName: rabbitmq
     * @Package: org.example.Work
     * @ClassName: consum1
     * @Description: java类作用描述
     * @Author: zhangq
     * @CreateDate: 2020-12-10 18:08
     * @UpdateUser: zhangq
     * @UpdateDate: 2020-12-10 18:08
     * @UpdateRemark: The modified content
     * @Version: 1.0 *
     */
    public class consum1 {
    
      public static void main(String[] args) throws IOException {
        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare("work",true,false,false,null);
        channel.basicConsume("work",true, new DefaultConsumer(channel){
          @Override
          public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
              byte[] body) throws IOException {
            System.out.println("消费者-1："+new String(body));
          }
        });
      }
    }

消费者2：

package org.example.Work;
    
    import com.rabbitmq.client.AMQP.BasicProperties;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.DefaultConsumer;
    import com.rabbitmq.client.Envelope;
    import java.io.IOException;
    import org.example.RabbitMQUtils;
    
    /**
     * java类简单作用描述
     *
     * @ProjectName: rabbitmq
     * @Package: org.example.Work
     * @ClassName: consum1
     * @Description: java类作用描述
     * @Author: zhangq
     * @CreateDate: 2020-12-10 18:08
     * @UpdateUser: zhangq
     * @UpdateDate: 2020-12-10 18:08
     * @UpdateRemark: The modified content
     * @Version: 1.0 *
     */
    public class consum2 {
    
      public static void main(String[] args) throws IOException {
        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare("work",true,false,false,null);
        channel.basicConsume("work",true, new DefaultConsumer(channel){
          @Override
          public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
              byte[] body) throws IOException {
            System.out.println("消费者-2："+new String(body));
          }
        });
      }
    }

顺序：先启动 2个消费者，在启动生产者看效果：（看出来是平均消费）

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1bmhlaWRhb2Rl_size_16_color_FFFFFF_t_70 2][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1bmhlaWRhb2Rl_size_16_color_FFFFFF_t_70 3][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1bmhlaWRhb2Rl_size_16_color_FFFFFF_t_70 4][]

看出来是平均消费的：

如果消费者1消费的慢，消费者2消费的快 ；则需要实现能者多劳模式：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1bmhlaWRhb2Rl_size_16_color_FFFFFF_t_70 5][]

.RabbitMQ中消息确认机制和能者多劳的实现：

消费者1：

package org.example.Work;
    
    import com.rabbitmq.client.AMQP.BasicProperties;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.DefaultConsumer;
    import com.rabbitmq.client.Envelope;
    import java.io.IOException;
    import org.example.RabbitMQUtils;
    
    /**
     * java类简单作用描述
     *
     * @ProjectName: rabbitmq
     * @Package: org.example.Work
     * @ClassName: consum1
     * @Description: java类作用描述
     * @Author: zhangq
     * @CreateDate: 2020-12-10 18:08
     * @UpdateUser: zhangq
     * @UpdateDate: 2020-12-10 18:08
     * @UpdateRemark: The modified content
     * @Version: 1.0 *
     */
    public class consum1 {
    
      public static void main(String[] args) throws IOException {
        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.basicQos(1);//每一次只能消费一个消息
        channel.queueDeclare("work", true, false, false, null);
        //autoAck 置位 false
        channel.basicConsume("work", false, new DefaultConsumer(channel) {
          @Override
          public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
              byte[] body) throws IOException {
            try {
              Thread.sleep(2000);
            } catch (InterruptedException e) {
              e.printStackTrace();
            }
            System.out.println("消费者-1：" + new String(body));
            //手动确认  false每次确认一个
            channel.basicAck(envelope.getDeliveryTag(),false);
          }
        });
      }
    }

消费者2：

package org.example.Work;
    
    import com.rabbitmq.client.AMQP.BasicProperties;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.DefaultConsumer;
    import com.rabbitmq.client.Envelope;
    import java.io.IOException;
    import org.example.RabbitMQUtils;
    
    /**
     * java类简单作用描述
     *
     * @ProjectName: rabbitmq
     * @Package: org.example.Work
     * @ClassName: consum1
     * @Description: java类作用描述
     * @Author: zhangq
     * @CreateDate: 2020-12-10 18:08
     * @UpdateUser: zhangq
     * @UpdateDate: 2020-12-10 18:08
     * @UpdateRemark: The modified content
     * @Version: 1.0 *
     */
    public class consum2 {
    
      public static void main(String[] args) throws IOException {
        Connection connection = RabbitMQUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.basicQos(1);//每一次只能消费一个消息
        channel.queueDeclare("work",true,false,false,null);
        channel.basicConsume("work",false, new DefaultConsumer(channel){
          @Override
          public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
              byte[] body) throws IOException {
            System.out.println("消费者-2："+new String(body));
            //手动确认  false每次确认一个
            channel.basicAck(envelope.getDeliveryTag(),false);
          }
        });
      }
    }

主要改动：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1bmhlaWRhb2Rl_size_16_color_FFFFFF_t_70 6][]

顺序：先启动 2个消费者，在启动生产者看效果：（看出来是不是平均消费）

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1bmhlaWRhb2Rl_size_16_color_FFFFFF_t_70 7][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1bmhlaWRhb2Rl_size_16_color_FFFFFF_t_70 8][]

这就实现了能者多劳

[https_github.com_heidaodageshiwo_ZqRabbitMq]: https://github.com/heidaodageshiwo/ZqRabbitMq
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1bmhlaWRhb2Rl_size_16_color_FFFFFF_t_70]: /images/20221120/afed20da04ed416cadbf336daa7bf8ed.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1bmhlaWRhb2Rl_size_16_color_FFFFFF_t_70 1]: /images/20221120/59e5b24bbbb5475f8b1d4930e240a7da.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1bmhlaWRhb2Rl_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20201210181623445.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1bmhlaWRhb2Rl,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1bmhlaWRhb2Rl_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20201210181627949.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1bmhlaWRhb2Rl,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1bmhlaWRhb2Rl_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/2020121018163486.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1bmhlaWRhb2Rl,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1bmhlaWRhb2Rl_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/20201210181849383.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1bmhlaWRhb2Rl,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1bmhlaWRhb2Rl_size_16_color_FFFFFF_t_70 6]: https://img-blog.csdnimg.cn/20201210182451110.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1bmhlaWRhb2Rl,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1bmhlaWRhb2Rl_size_16_color_FFFFFF_t_70 7]: https://img-blog.csdnimg.cn/20201210182541435.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1bmhlaWRhb2Rl,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1bmhlaWRhb2Rl_size_16_color_FFFFFF_t_70 8]: https://img-blog.csdnimg.cn/20201210182550406.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h1bmhlaWRhb2Rl,size_16,color_FFFFFF,t_70

SpringBoot整合Rabbitmq之===RabbitMQ的 Work 模型（无废话版）--------------------------------------------（三）

发表评论取消回复

还没有评论，来说两句吧...

相关阅读

相关 SpringBoot整合RocketMq之===RocketMq安装（无废话版）--------------------------------------------（一）

相关 SpringBoot整合Rabbitmq之===Rabbitmq集群搭建（无废话版）--------------------------------------------（八）

相关 SpringBoot整合Rabbitmq之===Springboot整合Rabbitmq（无废话版）--------------------------------------------（七）

相关 SpringBoot整合Rabbitmq之===RabbitMQ的 Routing之Topic（无废话版）--------------------------------------------（六）

相关 SpringBoot整合Rabbitmq之===RabbitMQ的 Routing之Dir模型（无废话版）--------------------------------------------（五）

相关 SpringBoot整合Rabbitmq之===RabbitMQ的 Fanout模型（无废话版）--------------------------------------------（四）

相关 SpringBoot整合Rabbitmq之===RabbitMQ的 Work 模型（无废话版）--------------------------------------------（三）

相关 SpringBoot整合Rabbitmq之===RabbitMQ的HelloWorld模型（无废话版）--------------------------------------------（二）

相关 SpringBoot整合Rabbitmq之===RabbitMQ的安装（无废话版）--------------------------------------------（一）

相关 RabbitMq入门（三）springBoot整合RabbitMQ