AI 助理
备案控制台
开发者社区 开发与运维 文章 正文

Java整合RabbitMQ实现生产消费（7种通讯方式）

2023-08-17 225
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： Java整合RabbitMQ实现生产消费（7种通讯方式）

环境说明

  • RabbitMQ环境，参考RabbitMQ环境搭建
  • Java版本：JDK1.8
  • Maven版本：apache-maven-3.6.3
  • 开发工具：IntelliJ IDEA

工程搭建

  1. 创建maven项目
  2. pom.xml文件引入RabbitMQ依赖
    <dependencies>
        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>5.16.0</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.13.2</version>
            <scope>compile</scope>
        </dependency>

    </dependencies>

连接RabbitMQ

import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class MQConnections {
   
   

    public static final String RABBITMQ_HOST = "127.0.0.1";
    public static final int RABBITMQ_PORT = 5672;
    public static final String RABBITMQ_USERNAME = "guest";
    public static final String RABBITMQ_PASSWORD = "guest";
    public static final String RABBITMQ_VIRTUAL_HOST = "/";

    /**
     * 构建RabbitMQ连接对象
     *
     * @return
     */
    public static Connection getConnection() throws IOException, TimeoutException {
   
   
        //1.创建Connection工厂
        ConnectionFactory factory = new ConnectionFactory();
        //2.设置Rabbitmq连接信息
        factory.setHost(RABBITMQ_HOST);
        factory.setPort(RABBITMQ_PORT);
        factory.setUsername(RABBITMQ_USERNAME);
        factory.setPassword(RABBITMQ_PASSWORD);
        factory.setVirtualHost(RABBITMQ_VIRTUAL_HOST);
        //3.返回连接对象
        return factory.newConnection();
    }

}

通讯模式

1.简单通讯

即一个生产者可以向一个队列发送消息，一个消费者可以尝试从一个队列接收数据。如下图：
在这里插入图片描述

    public final static String HELLO_QUEUE_NAME = "hello";
    @Test
    public void publish_hello() throws IOException, TimeoutException {
   
   
        //1.获取连接对象
        Connection connection = MQConnections.getConnection();
        //2.构建Channl
        Channel channel = connection.createChannel();
        //3.构建队列,queueDeclare("队列名称","是否持久化队列","是否只允许一个队列消费","长时间未使用是否删除"，"其他参数")
        channel.queueDeclare(HELLO_QUEUE_NAME, false, false, false, null);
        //4.发布消息
        String msg = "hello,world";
        channel.basicPublish("", HELLO_QUEUE_NAME, null, msg.getBytes());
    }
    @Test
    public void consume_hello() throws IOException, TimeoutException {
   
   
        //1.获取连接对象
        Connection connection = MQConnections.getConnection();
        //2.构建Channl
        Channel channel = connection.createChannel();
        //3.构建队列,queueDeclare("队列名称","是否持久化队列","是否只允许一个队列消费","长时间未使用是否删除"，"其他参数")
        channel.queueDeclare(HELLO_QUEUE_NAME, false, false, false, null);
        //4.监听消息
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
   
   
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("Received '" + message + "'");
        };
        channel.basicConsume(HELLO_QUEUE_NAME, true, deliverCallback, consumerTag -> {
   
   
        });
        System.in.read();
    }

2.工作队列通讯

与简单通讯一样，当消费能力不足或想要提高吞吐时可添加多个消费者进行处理业务。如下图，队列中的消息会逐条被C1和C2消费。
在这里插入图片描述

    public final static String WORK_QUEUE_NAME = "work";

    @Test
    public void publish_work_queue() throws IOException, TimeoutException {
   
   
        //1.获取连接对象
        Connection connection = MQConnections.getConnection();
        //2.构建Channl
        Channel channel = connection.createChannel();
        //3.构建队列,queueDeclare("队列名称","是否持久化队列","是否只允许一个队列消费","长时间未使用是否删除"，"其他参数")
        channel.queueDeclare(WORK_QUEUE_NAME, false, false, false, null);
        //4.发布消息
        String msg = "hello,work queue";
        channel.basicPublish("", WORK_QUEUE_NAME, null, msg.getBytes());
    }
        @Test
    public void consume_work_queue1() throws IOException, TimeoutException {
   
   
        //1.获取连接对象
        Connection connection = MQConnections.getConnection();
        //2.构建Channl
        Channel channel = connection.createChannel();
        //3.构建队列,queueDeclare("队列名称","是否持久化队列","是否只允许一个队列消费","长时间未使用是否删除"，"其他参数")
        channel.queueDeclare(WORK_QUEUE_NAME, false, false, false, null);
        //4.监听消息
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
   
   
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("consume1 Received '" + message + "'");
        };
        channel.basicConsume(WORK_QUEUE_NAME, true, deliverCallback, consumerTag -> {
   
   
        });
        System.in.read();
    }

    @Test
    public void consume_work_queue2() throws IOException, TimeoutException {
   
   
        //1.获取连接对象
        Connection connection = MQConnections.getConnection();
        //2.构建Channl
        Channel channel = connection.createChannel();
        //3.构建队列,queueDeclare("队列名称","是否持久化队列","是否只允许一个队列消费","长时间未使用是否删除"，"其他参数")
        channel.queueDeclare(WORK_QUEUE_NAME, false, false, false, null);
        //4.监听消息
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
   
   
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("consume2 Received '" + message + "'");
        };
        channel.basicConsume(WORK_QUEUE_NAME, true, deliverCallback, consumerTag -> {
   
   
        });
        System.in.read();
    }

3.发布/订阅通讯

工作队列背后的假设是，每个任务只交付给一个消费者做同一件事。如果要交付给多个消费者做不同的事，需要引入交换机实现一个完整的消息传递模型，这种模式被称为“发布/订阅”。如下图，消息会发布到交换机中，交换机向绑定的队列同时发送消息，最终C1和C2会同时消费此条消息。
在这里插入图片描述

public final static String PUB_EXCHANGE_NAME = "pub-ex";
    public final static String PUB1_QUEUE_NAME = "pub-que1";
    public final static String PUB2_QUEUE_NAME = "pub-que2";

    @Test
    public void publish_pub_sub() throws IOException, TimeoutException {
   
   
        //1.获取连接对象
        Connection connection = MQConnections.getConnection();
        //2.构建Channl
        Channel channel = connection.createChannel();
        //3.构建交换机
        channel.exchangeDeclare(PUB_EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
        //4.构建队列,queueDeclare("队列名称","是否持久化队列","是否只允许一个队列消费","长时间未使用是否删除"，"其他参数")
        channel.queueDeclare(PUB1_QUEUE_NAME, false, false, false, null);
        channel.queueDeclare(PUB2_QUEUE_NAME, false, false, false, null);
        //5.绑定队列
        channel.queueBind(PUB1_QUEUE_NAME, PUB_EXCHANGE_NAME, "");
        channel.queueBind(PUB2_QUEUE_NAME, PUB_EXCHANGE_NAME, "");
        //6.发布消息
        String msg = "hello,pub/sub";
        channel.basicPublish(PUB_EXCHANGE_NAME, "", null, msg.getBytes());
    }
    @Test
    public void consume_pub_sub1() throws IOException, TimeoutException {
   
   
        //1.获取连接对象
        Connection connection = MQConnections.getConnection();
        //2.构建Channl
        Channel channel = connection.createChannel();
        //3.构建队列,queueDeclare("队列名称","是否持久化队列","是否只允许一个队列消费","长时间未使用是否删除"，"其他参数")
        channel.queueDeclare(PUB1_QUEUE_NAME, false, false, false, null);
        //4.监听消息
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
   
   
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("pub_sub1 Received '" + message + "'");
        };
        channel.basicConsume(PUB1_QUEUE_NAME, true, deliverCallback, consumerTag -> {
   
   
        });
        System.in.read();
    }

    @Test
    public void consume_pub_sub2() throws IOException, TimeoutException {
   
   
        //1.获取连接对象
        Connection connection = MQConnections.getConnection();
        //2.构建Channl
        Channel channel = connection.createChannel();
        //3.构建队列,queueDeclare("队列名称","是否持久化队列","是否只允许一个队列消费","长时间未使用是否删除"，"其他参数")
        channel.queueDeclare(PUB2_QUEUE_NAME, false, false, false, null);
        //4.监听消息
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
   
   
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("pub_sub2 Received '" + message + "'");
        };
        channel.basicConsume(PUB2_QUEUE_NAME, true, deliverCallback, consumerTag -> {
   
   
        });
        System.in.read();
    }

4.路由通讯

发布/订阅模式是交换机将一条消息同时路由给多个队列，“路由”模式可以将消息通过交换机指定到某个队列中从而被消费。如下图，交换机将所有类型的日志路由到一个队列中，将error类型的日志路由到另一个队列中。
在这里插入图片描述

    public final static String ROUT_EXCHANGE_NAME = "rout-ex";
    public final static String ROUTALL_QUEUE_NAME = "rout-queall";
    public final static String ROUTONE_QUEUE_NAME = "rout-queone";

    @Test
    public void publish_routing() throws IOException, TimeoutException {
   
   
        //1.获取连接对象
        Connection connection = MQConnections.getConnection();
        //2.构建Channl
        Channel channel = connection.createChannel();
        //3.构建交换机
        channel.exchangeDeclare(ROUT_EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
        //4.构建队列,queueDeclare("队列名称","是否持久化队列","是否只允许一个队列消费","长时间未使用是否删除"，"其他参数")
        channel.queueDeclare(ROUTALL_QUEUE_NAME, false, false, false, null);
        channel.queueDeclare(ROUTONE_QUEUE_NAME, false, false, false, null);
        //5.绑定队列
        channel.queueBind(ROUTALL_QUEUE_NAME, ROUT_EXCHANGE_NAME, "all");
        channel.queueBind(ROUTONE_QUEUE_NAME, ROUT_EXCHANGE_NAME, "one");
        //6.发布消息
        String msg1 = "hello,1-all";
        String msg2 = "hello,2-all";
        String msg3 = "hello,1-one";
        channel.basicPublish(ROUT_EXCHANGE_NAME, "all", null, msg1.getBytes());
        channel.basicPublish(ROUT_EXCHANGE_NAME, "all", null, msg2.getBytes());
        channel.basicPublish(ROUT_EXCHANGE_NAME, "one", null, msg3.getBytes());
    }
    @Test
    public void consume_routing_all() throws IOException, TimeoutException {
   
   
        //1.获取连接对象
        Connection connection = MQConnections.getConnection();
        //2.构建Channl
        Channel channel = connection.createChannel();
        //3.构建队列,queueDeclare("队列名称","是否持久化队列","是否只允许一个队列消费","长时间未使用是否删除"，"其他参数")
        channel.queueDeclare(ROUTALL_QUEUE_NAME, false, false, false, null);
        //4.监听消息
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
   
   
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("consume_routing_all Received '" + message + "'");
        };
        channel.basicConsume(ROUTALL_QUEUE_NAME, true, deliverCallback, consumerTag -> {
   
   
        });
        System.in.read();
    }

    @Test
    public void consume_routing_one() throws IOException, TimeoutException {
   
   
        //1.获取连接对象
        Connection connection = MQConnections.getConnection();
        //2.构建Channl
        Channel channel = connection.createChannel();
        //3.构建队列,queueDeclare("队列名称","是否持久化队列","是否只允许一个队列消费","长时间未使用是否删除"，"其他参数")
        channel.queueDeclare(ROUTONE_QUEUE_NAME, false, false, false, null);
        //4.监听消息
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
   
   
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("consume_routing_one Received '" + message + "'");
        };
        channel.basicConsume(ROUTONE_QUEUE_NAME, true, deliverCallback, consumerTag -> {
   
   
        });
        System.in.read();
    }

5.主题通讯

“路由”模式仍然有局限性——它不能基于多个标准进行路由。主题可以带来很大的灵活性，发送到主题交换的消息不能有任意的routing_key，它必须是一个用点分隔的单词列表，routing_key有两种重要的特殊情况：

  • *只能代替一个词。

  • 可以替换零个或多个单词。

    在这里插入图片描述

      public final static String TOPIC_EXCHANGE_NAME = "topic-ex";
  public final static String TOPICALL_QUEUE_NAME = "topic-queall";
  public final static String TOPICONE_QUEUE_NAME = "topic-queone";

  @Test
  public void publish_topic() throws IOException, TimeoutException {
     
     
      //1.获取连接对象
      Connection connection = MQConnections.getConnection();
      //2.构建Channl
      Channel channel = connection.createChannel();
      //3.构建交换机
      channel.exchangeDeclare(TOPIC_EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
      //4.构建队列,queueDeclare("队列名称","是否持久化队列","是否只允许一个队列消费","长时间未使用是否删除"，"其他参数")
      channel.queueDeclare(TOPICALL_QUEUE_NAME, false, false, false, null);
      channel.queueDeclare(TOPICONE_QUEUE_NAME, false, false, false, null);
      //5.绑定队列
      channel.queueBind(TOPICALL_QUEUE_NAME, TOPIC_EXCHANGE_NAME, "*.all.*");
      channel.queueBind(TOPICONE_QUEUE_NAME, TOPIC_EXCHANGE_NAME, "#.one");
      //6.发布消息
      String msg1 = "hello.all.world";
      String msg2 = "hello.world.one";
      channel.basicPublish(TOPIC_EXCHANGE_NAME, "hello.all.world", null, msg1.getBytes());
      channel.basicPublish(TOPIC_EXCHANGE_NAME, "hello.world.one", null, msg2.getBytes());
  }
      @Test
  public void consume_topic_all() throws IOException, TimeoutException {
     
     
      //1.获取连接对象
      Connection connection = MQConnections.getConnection();
      //2.构建Channl
      Channel channel = connection.createChannel();
      //3.构建队列,queueDeclare("队列名称","是否持久化队列","是否只允许一个队列消费","长时间未使用是否删除"，"其他参数")
      channel.queueDeclare(TOPICALL_QUEUE_NAME, false, false, false, null);
      //4.监听消息
      DeliverCallback deliverCallback = (consumerTag, delivery) -> {
     
     
          String message = new String(delivery.getBody(), "UTF-8");
          System.out.println("consume_topic_all Received '" + message + "'");
      };
      channel.basicConsume(TOPICALL_QUEUE_NAME, true, deliverCallback, consumerTag -> {
     
     
      });
      System.in.read();
  }

  @Test
  public void consume_topic_one() throws IOException, TimeoutException {
     
     
      //1.获取连接对象
      Connection connection = MQConnections.getConnection();
      //2.构建Channl
      Channel channel = connection.createChannel();
      //3.构建队列,queueDeclare("队列名称","是否持久化队列","是否只允许一个队列消费","长时间未使用是否删除"，"其他参数")
      channel.queueDeclare(TOPICONE_QUEUE_NAME, false, false, false, null);
      //4.监听消息
      DeliverCallback deliverCallback = (consumerTag, delivery) -> {
     
     
          String message = new String(delivery.getBody(), "UTF-8");
          System.out.println("consume_topic_one Received '" + message + "'");
      };
      channel.basicConsume(TOPICONE_QUEUE_NAME, true, deliverCallback, consumerTag -> {
     
     
      });
      System.in.read();
  }

    6.RPC通讯

    RabbitMQ作为消息中间件可以达到应用解耦效果，如果想达到RPC远程调用同步返回结果，RabbitMQ同样支持，其原理如下：

  • 发布者发送消息时指定一个回调队列和唯一id
  • 消费者处理完成后将结果发送到回调队列中
  • 发布者按照唯一id接收消息并处理

如下图
在这里插入图片描述

    public final static String RPC_QUEUE_NAME = "rpc-que";
    public final static String RPCCALLBACK_QUEUE_NAME = "rpc-callback-que";

    @Test
    public void publish_rpc() throws IOException, TimeoutException {
   
   
        //1.获取连接对象
        Connection connection = MQConnections.getConnection();
        //2.构建Channl
        Channel channel = connection.createChannel();
        //3.构建队列,queueDeclare("队列名称","是否持久化队列","是否只允许一个队列消费","长时间未使用是否删除"，"其他参数")
        channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);
        // replyTo回调队列
        channel.queueDeclare(RPCCALLBACK_QUEUE_NAME, false, false, false, null);
        //4.发布消息
        String msg = "hello rpc";
        String correlationId = UUID.randomUUID().toString();
        /*AMQP 协议预先定义了一组与消息一起使用的14个属性。除了以下属性外，大多数属性很少使用：
        deliveryMode：将消息标记为持久（值为2）或瞬时（任何其他值）。
        contentType：用于描述编码的mime类型。例如，对于常用的JSON编码，最好将此属性设置为：application/JSON。
        replyTo：通常用于命名回调队列。
        correlationId：用于将RPC响应与请求关联。*/
        AMQP.BasicProperties basicProperties = new AMQP.BasicProperties.Builder().replyTo(RPCCALLBACK_QUEUE_NAME).correlationId(correlationId).build();
        //5.回调响应结果
        channel.basicPublish("", RPC_QUEUE_NAME, basicProperties, msg.getBytes());
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
   
   
            String message = new String(delivery.getBody(), "UTF-8");
            String recid = delivery.getProperties().getCorrelationId();
            if (correlationId.equalsIgnoreCase(recid)) System.out.println("rpc-callback-que   '" + message + "'");
        };
        channel.basicConsume(RPCCALLBACK_QUEUE_NAME, true, deliverCallback, consumerTag -> {
   
   
        });
        System.in.read();
    }
        @Test
    public void consume_rpc() throws IOException, TimeoutException {
   
   
        //1.获取连接对象
        Connection connection = MQConnections.getConnection();
        //2.构建Channl
        Channel channel = connection.createChannel();
        //3.构建队列,queueDeclare("队列名称","是否持久化队列","是否只允许一个队列消费","长时间未使用是否删除"，"其他参数")
        channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);
        //4.监听消息
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
   
   
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("consume_rpc Received '" + message + "'");
            String correlationId = delivery.getProperties().getCorrelationId();
            AMQP.BasicProperties basicProperties = new AMQP.BasicProperties.Builder().correlationId(correlationId).build();
            String replyTo = delivery.getProperties().getReplyTo();
            String callbackmsg = "rpc callback";
            channel.basicPublish("", replyTo, basicProperties, callbackmsg.getBytes());
        };
        channel.basicConsume(RPC_QUEUE_NAME, true, deliverCallback, consumerTag -> {
   
   
        });
        System.in.read();
    }

7.Publisher确认通讯

Publisher确认是RabbitMQ扩展以实现可靠发布。当在通道上启用发布者确认时，代理将异步确认客户端发布的消息，这意味着它们已在服务器端得到处理。

    public final static String CONFIRM_EXCHANGE_NAME = "confirm-ex";
    public final static String CONFIRM_QUEUE_NAME = "confirm-que";

    @Test
    public void publish_confirm() throws IOException, TimeoutException {
   
   
        //1.获取连接对象
        Connection connection = MQConnections.getConnection();
        //2.构建Channl
        Channel channel = connection.createChannel();
        //3.开启确认选项
        channel.confirmSelect();
        //4.构建交换机
        channel.exchangeDeclare(CONFIRM_EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
        //5.构建队列,queueDeclare("队列名称","是否持久化队列","是否只允许一个队列消费","长时间未使用是否删除"，"其他参数")
        channel.queueDeclare(CONFIRM_QUEUE_NAME, true, false, false, null);
        //6.绑定队列
        String right_routing_key = "confirm";
        String error_routing_key = "confirm_err";
        channel.queueBind(CONFIRM_QUEUE_NAME, CONFIRM_EXCHANGE_NAME, right_routing_key);
        //7.消息到达交换机确认监听
        channel.addConfirmListener((sequenceNumber, multiple) -> {
   
   
            System.out.println("消息成功发送到交换机");
        }, (sequenceNumber, multiple) -> {
   
   
            System.err.println("消息未发送到交换机，补偿操作。");
        });
        //8.消息到达队列确认监听
        channel.addReturnListener((replyCode, replyText, exchange, routingKey, basicProperties, body) -> {
   
   
            System.err.format("消息 %s 未路由到指定队列: %s, replyText: %s,replyCode: %d%n", body, routingKey, replyText, replyCode);
        });
        //设置消息持久化
        AMQP.BasicProperties basicProperties = new AMQP.BasicProperties.Builder().deliveryMode(2).build();
        //7.发布消息
        String msg = "hello confirm";
        channel.basicPublish(CONFIRM_EXCHANGE_NAME, error_routing_key,true, basicProperties, msg.getBytes());
        System.in.read();
    }

    @Test
    public void consume_ack() throws IOException, TimeoutException {
   
   
        //1.获取连接对象
        Connection connection = MQConnections.getConnection();
        //2.构建Channl
        Channel channel = connection.createChannel();
        //3.构建队列,queueDeclare("队列名称","是否持久化队列","是否只允许一个队列消费","长时间未使用是否删除"，"其他参数")
        channel.queueDeclare(CONFIRM_QUEUE_NAME, true, false, false, null);
        //4.监听消息
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
   
   
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("consume_routing_one Received '" + message + "'");
            //消息处理后手动ACK
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
        };
        // ack为false
        channel.basicConsume(CONFIRM_QUEUE_NAME, false, deliverCallback, consumerTag -> {
   
   
        });
        System.in.read();
    }

代码仓库

https://gitee.com/codeWBG/learn_rabbitmq

文章标签：
云消息队列 MQ
消息中间件
Java
Maven
开发工具
关键词：
Java通讯
云消息队列 MQ java
Java生产
java云消息队列 MQ
Java RabbitMQ
相关实践学习
消息队列RocketMQ版：基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能，涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列：消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验：基于MNS，0基础轻松构建 Web Client 本节课带您一起基于MNS，0基础轻松构建 Web Client
王二蛋呀
目录
相关文章
三分钟热度的鱼
|
2月前
|
消息中间件 网络协议 RocketMQ
消息队列 MQ产品使用合集之broker开启proxy，启动之后producer生产消息始终都只到一个broker，该怎么办
消息队列（MQ）是一种用于异步通信和解耦的应用程序间消息传递的服务，广泛应用于分布式系统中。针对不同的MQ产品，如阿里云的RocketMQ、RabbitMQ等，它们在实现上述场景时可能会有不同的特性和优势，比如RocketMQ强调高吞吐量、低延迟和高可用性，适合大规模分布式系统；而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集，这些场景涵盖了多种行业和业务需求。
三分钟热度的鱼
47 1
yuanzhengme
|
1天前
|
消息中间件 负载均衡 Java
Java微服务通讯方式有哪些？
【8月更文挑战第18天】Java微服务通讯方式有哪些？
yuanzhengme
9 1
VipSoft
|
5天前
|
Java
MQTT(EMQX) - Java 调用 MQTT Demo 代码
MQTT(EMQX) - Java 调用 MQTT Demo 代码
VipSoft
8 0
MQTT(EMQX) - Java 调用 MQTT Demo 代码
蓝易云
|
11天前
|
设计模式 安全 Java
Java多线程通讯
这些机制都是多线程通信的关键，使用它们可以在Java中实现高效、安全的线程协作。在实现多线程通信时，需注意死锁、饥饿、竞态条件等多线程问题。确保对多线程的理解深入和正确的设计模式应用将会是编写稳健多线程程序的基础。
蓝易云
7 0
weixin_836869520
|
1月前
|
消息中间件 Java Maven
如何在Java中使用RabbitMQ
如何在Java中使用RabbitMQ
weixin_836869520
96 2
杰哥在此
|
1月前
|
存储 设计模式 监控
Java面试题：如何在不牺牲性能的前提下，实现一个线程安全的单例模式？如何在生产者-消费者模式中平衡生产和消费的速度？Java内存模型规定了变量在内存中的存储和线程间的交互规则
Java面试题：如何在不牺牲性能的前提下，实现一个线程安全的单例模式？如何在生产者-消费者模式中平衡生产和消费的速度？Java内存模型规定了变量在内存中的存储和线程间的交互规则
杰哥在此
30 0
yqd666
|
1月前
|
消息中间件 负载均衡 Java
JAVA面试之MQ
JAVA面试之MQ
yqd666
42 0
长梦
|
2月前
|
消息中间件 Java
Java一分钟之-RabbitMQ：AMQP协议实现
【6月更文挑战第11天】RabbitMQ是基于AMQP协议的开源消息队列服务，支持多种消息模式。本文介绍了RabbitMQ的核心概念：生产者、消费者、交换器、队列和绑定，以及常见问题和解决方案。例如，通过设置消息持久化和确认机制防止消息丢失，配置死信队列处理不可消费消息，以及妥善管理资源防止泄漏。还提供了Java代码示例，帮助读者理解和使用RabbitMQ。通过理解这些基础和最佳实践，可以提升RabbitMQ在分布式系统中的可靠性和效率。
长梦
56 0
Java一分钟之-RabbitMQ：AMQP协议实现
米诺斯
|
2月前
|
Java
java使用Supplier接口的get生产一个数据
java使用Supplier接口的get生产一个数据
米诺斯
16 0
米诺斯
|
2月前
|
Java
java使用ServerSocket和Socket实现客户端与服务端通讯
java使用ServerSocket和Socket实现客户端与服务端通讯
米诺斯
21 0

热门文章

最新文章

  • 1
    信用算力基于 RocketMQ 实现金融级数据服务的实践
  • 2
    springboot+redis+rabbitmq实现模拟秒杀系统(附带docker安装mysql，rabbitmq，redis教程)
  • 3
    WebAPi的可视化输出模式(RabbitMQ、消息补偿相关)所有webapi似乎都缺失的一个功能
  • 4
    RabbitMQ 消费端限流、TTL、死信队列
  • 5
    python rabbitmq send messages
  • 6
    MQ是如何保证消息不丢失的,你这样回答面试官一定说I Want U
  • 7
    RocketMQ 5.0: 存储计算分离新思路
  • 8
    RocketMQ在面试中那些常见问题及答案+汇总
  • 9
    RabbitMQ入门指南：初学者也能读懂的教程（一）
  • 10
    rabbit.bat server启动闪退，rabbitmq (ArgumentError) argument error，rabbitmq页面无法开启解决方案
  • 1
    Java并发编程：理解并应用ReentrantLock
    42
  • 2
    Java并发编程：深入理解线程池
    24
  • 3
    Java 8新特性解析及应用区块链技术在供应链管理中的应用与挑战
    39
  • 4
    Java 效率编码 必备插件 Lombok 让代码更优雅
    20
  • 5
    Java代码一键生成数据库文档（案例详解）
    124
  • 6
    java一行代码实现RESTFul接口
    41
  • 7
    Java远程连接本地开源分布式搜索引擎ElasticSearch
    31
  • 8
    Spring Boot单元测试报错java.lang.IllegalStateException: Could not load TestContextBootstrapper [null]
    79
  • 9
    Java并发编程：深入理解线程池
    31
  • 10
    Java中 a+=b和a=a+b有什么区别？
    25

    • 相关课程

    更多
  • 消息队列 RocketMQ 消息集成
  • 消息队列 MNS 入门课程
  • RocketMQ知识精讲与项目实战（第一阶段）
  • RocketMQ知识精讲与项目实战（第二阶段）
  • RocketMQ知识精讲与项目实战（第三阶段）
  • 微服务实战-RocketMQ Binder

    • 相关电子书

    更多
  • RocketMQ Client-GO 介绍
  • RocketMQ Prometheus Exporter 打造定制化 DevOps 平台
  • 基于 RocketMQ Prometheus Exporter 打造定制化 DevOps 平台

    • 相关实验场景

    更多
  • 快速学会使用云消息队列RabbitMQ版
  • 消息队列RocketMQ版：消费异常运维排查体验
  • 消息队列RocketMQ版：定时消息通知功能体验
  • 阿里云平台上进行Java程序的编译与运行
  • 消息队列RocketMQ版：基础消息收发功能体验
  • 使用Java面向对象编写网络通信程序应用
    • 下一篇
    通义千问API入门教程