消息中间件知识梳理

消息中间件

为什么要使用MQ？

• 实现异步通信
• 实现系统解耦
• 实现流量削峰
• 实现广播通信

MQ带来的问题

• 运维成本的增加
• 系统的可用性降低
• 系统复杂性提高

AMQP协议是应用层的协议

RabbitMQ

因为是Erlang编写的，而且Erlang是为电话交换机编写的语言，天生适合分布式和高并发。

默认端口是5672，RabbitMq服务器我们叫做Broker。消费者可生产者都要跟broker建立连接，这个连接属于TCP的长连接。

channel

为了节省性能损耗，节省时间。在AMQP里面引入了Channel的概念，他是一个虚拟的连接。也就是通道或者消息信道。不同的Channel是相互隔离的，每个Channel都有自己的编号。每个客户端线程都有自己的Channel。

Queue

在Broker上有一个对象用来存储消息，在RabbitMQ里面这个对象叫做Queue。

Consumer

消费者消费消息有两种模式。

一种是pull模式，对应的方法是basicGet。消息存放在服务端，只有消费者主动获取才能拿到消息。

另一种是push模式，对应的方法是basicConsume，只要生产者发消息到服务器，就马上推送给消费者，消息保存在客户端，实时性很高，如果消费者不过来会造成消息积压。

Exchange

是帮助路由消息的组件。不管有多少个队列需要接收消息，只需要发送到Exchange就可以，由它来分发。Exchange不会存储消息，它只做一件事，根据规则分发消息。

Exchange和这些需要接收消息的队列必须建立一个绑定关系，并且为每个队列指定一个特殊的标识。Exchange和队列是多对多的绑定关系，一个交换机的消息一个路由给多个队列，一个队列也可以接收来自多个交换机的消息。

Vhost

不同的的Vhost可以实现资源的隔离和权限的控制。不同的Vhost中可以由同名的Exchange和Queue。默认的Vhost名字是”/“。

路由方式

一共有四种类型的交换机，Direct、Topic、Fanout、Headers。Headers不常用。

Direct直连

一个队列与直连类型的交换机绑定，需指定一个明确的绑定键（binding key）。生产者发送消息时会携带一个路由键（routing key）。

当消息的路由键与某个队列的绑定键完全匹配时，这条消息擦灰从交换机路由到这个队列上。多个队列也可以使用相同的绑定键。

public class MyConsumer {
    private final static String EXCHANGE_NAME = "SIMPLE_EXCHANGE";
    private final static String QUEUE_NAME = "SIMPLE_QUEUE";

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        // 连接IP
        factory.setHost("192.168.242.110");
        // 默认监听端口
        factory.setPort(5672);
        // 虚拟机
        factory.setVirtualHost("/");

        // 设置访问的用户
        factory.setUsername("admin");
        factory.setPassword("admin");
        // 建立连接
        Connection conn = factory.newConnection();
        // 创建消息通道
        Channel channel = conn.createChannel();

        // 声明交换机
        // String exchange, String type, boolean durable, boolean autoDelete, Map<String, Object> arguments
        channel.exchangeDeclare(EXCHANGE_NAME,"direct",false, false, null);

        // 声明队列
        // String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        System.out.println(" Waiting for message....");

        // 绑定队列和交换机
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"gupao.best");

        // 创建消费者
        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("Received message : '" + msg + "'");
                System.out.println("consumerTag : " + consumerTag );
                System.out.println("deliveryTag : " + envelope.getDeliveryTag() );
            }
        };

        // 开始获取消息
        // String queue, boolean autoAck, Consumer callback
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

public class MyProducer {
    private final static String EXCHANGE_NAME = "SIMPLE_EXCHANGE";

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        // 连接IP
        factory.setHost("192.168.242.110");
        // 连接端口
        factory.setPort(5672);
        // 虚拟机
        factory.setVirtualHost("/");
        // 用户
        factory.setUsername("admin");
        factory.setPassword("admin");

        // 建立连接
        Connection conn = factory.newConnection();
        // 创建消息通道
        Channel channel = conn.createChannel();

        // 发送消息
        String msg = "Hello world, Rabbit MQ";

        // String exchange, String routingKey, BasicProperties props, byte[] body
        channel.basicPublish(EXCHANGE_NAME, "gupao.best", null, msg.getBytes());

        channel.close();
        conn.close();
    }
}

Topic主题

一个队列与主题类型的交换机绑定时，可以在绑定键中使用通配符。支持两个通配符：

• #代表0个或多个单词

• *代表不多不少一个单词

单词用代表用英文的.分开的字符。例如a.bc.def是三个单词。

Fanout广播

广播类型的交换机与队列绑定时，不需要指定绑定键。消息到达交换机时，所有与之绑定了的队列，都会收到相同的消息副本。

springmvc用法

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
     http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
     http://www.springframework.org/schema/rabbit
     http://www.springframework.org/schema/rabbit/spring-rabbit-1.2.xsd">

    <!--配置connection-factory，指定连接rabbit server参数 -->
    <rabbit:connection-factory id="connectionFactory" virtual-host="/" username="admin" password="admin" host="192.168.242.110" port="5672" />

    <!--通过指定下面的admin信息，当前producer中的exchange和queue会在rabbitmq服务器上自动生成 -->
    <rabbit:admin id="connectAdmin" connection-factory="connectionFactory" />

    <!--######分隔线######-->
    <!--定义queue -->
    <rabbit:queue name="MY_FIRST_QUEUE" durable="true" auto-delete="false" exclusive="false" declared-by="connectAdmin" />

    <!--定义direct exchange，绑定MY_FIRST_QUEUE -->
    <rabbit:direct-exchange name="MY_DIRECT_EXCHANGE" durable="true" auto-delete="false" declared-by="connectAdmin">
        <rabbit:bindings>
            <rabbit:binding queue="MY_FIRST_QUEUE" key="FirstKey">
            </rabbit:binding>
        </rabbit:bindings>
    </rabbit:direct-exchange>

    <!--定义rabbit template用于数据的接收和发送 -->
    <rabbit:template id="amqpTemplate" connection-factory="connectionFactory" exchange="MY_DIRECT_EXCHANGE" />

    <!--消息接收者 -->
    <bean id="messageReceiver" class="com.gupaoedu.consumer.FirstConsumer"></bean>

    <!--queue listener 观察 监听模式 当有消息到达时会通知监听在对应的队列上的监听对象 -->
    <rabbit:listener-container connection-factory="connectionFactory">
        <rabbit:listener queues="MY_FIRST_QUEUE" ref="messageReceiver" />
    </rabbit:listener-container>




    <!--定义queue -->
    <rabbit:queue name="MY_SECOND_QUEUE" durable="true" auto-delete="false" exclusive="false" declared-by="connectAdmin" />

    <!-- 将已经定义的Exchange绑定到MY_SECOND_QUEUE，注意关键词是key -->
    <rabbit:direct-exchange name="MY_DIRECT_EXCHANGE" durable="true" auto-delete="false" declared-by="connectAdmin">
        <rabbit:bindings>
            <rabbit:binding queue="MY_SECOND_QUEUE" key="SecondKey"></rabbit:binding>
        </rabbit:bindings>
    </rabbit:direct-exchange>

    <!-- 消息接收者 -->
    <bean id="receiverSecond" class="com.gupaoedu.consumer.SecondConsumer"></bean>

    <!-- queue litener 观察 监听模式 当有消息到达时会通知监听在对应的队列上的监听对象 -->
    <rabbit:listener-container connection-factory="connectionFactory">
        <rabbit:listener queues="MY_SECOND_QUEUE" ref="receiverSecond" />
    </rabbit:listener-container>

    <!--######分隔线######-->
    <!--定义queue -->
    <rabbit:queue name="MY_THIRD_QUEUE" durable="true" auto-delete="false" exclusive="false" declared-by="connectAdmin" />

    <!-- 定义topic exchange，绑定MY_THIRD_QUEUE，注意关键词是pattern -->
    <rabbit:topic-exchange name="MY_TOPIC_EXCHANGE" durable="true" auto-delete="false" declared-by="connectAdmin">
        <rabbit:bindings>
            <rabbit:binding queue="MY_THIRD_QUEUE" pattern="#.Third.#"></rabbit:binding>
        </rabbit:bindings>
    </rabbit:topic-exchange>

    <!--定义rabbit template用于数据的接收和发送 -->
    <rabbit:template id="amqpTemplate2" connection-factory="connectionFactory" exchange="MY_TOPIC_EXCHANGE" />

    <!-- 消息接收者 -->
    <bean id="receiverThird" class="com.gupaoedu.consumer.ThirdConsumer"></bean>

    <!-- queue litener 观察 监听模式 当有消息到达时会通知监听在对应的队列上的监听对象 -->
    <rabbit:listener-container connection-factory="connectionFactory">
        <rabbit:listener queues="MY_THIRD_QUEUE" ref="receiverThird" />
    </rabbit:listener-container>

    <!--######分隔线######-->
    <!--定义queue -->
    <rabbit:queue name="MY_FOURTH_QUEUE" durable="true" auto-delete="false" exclusive="false" declared-by="connectAdmin" />

    <!-- 定义fanout exchange，绑定MY_FIRST_QUEUE 和 MY_FOURTH_QUEUE -->
    <rabbit:fanout-exchange name="MY_FANOUT_EXCHANGE" auto-delete="false" durable="true" declared-by="connectAdmin" >
        <rabbit:bindings>
            <rabbit:binding queue="MY_FIRST_QUEUE"></rabbit:binding>
            <rabbit:binding queue="MY_FOURTH_QUEUE"></rabbit:binding>
        </rabbit:bindings>
    </rabbit:fanout-exchange>

    <!-- 消息接收者 -->
    <bean id="receiverFourth" class="com.gupaoedu.consumer.FourthConsumer"></bean>

    <!-- queue litener 观察 监听模式 当有消息到达时会通知监听在对应的队列上的监听对象 -->
    <rabbit:listener-container connection-factory="connectionFactory">
        <rabbit:listener queues="MY_FOURTH_QUEUE" ref="receiverFourth" />
    </rabbit:listener-container>
</beans>

@Service
public class MessageProducer {
    private Logger logger = LoggerFactory.getLogger(MessageProducer.class);

    @Autowired
    @Qualifier("amqpTemplate")
    private AmqpTemplate amqpTemplate;

    @Autowired
    @Qualifier("amqpTemplate2")
    private AmqpTemplate amqpTemplate2;

    /**
     * 演示三种交换机的使用
     *
     * @param message
     */
    public void sendMessage(Object message) {


        // amqpTemplate 默认交换机 MY_DIRECT_EXCHANGE
        // amqpTemplate2 默认交换机 MY_TOPIC_EXCHANGE

        // Exchange 为 direct 模式，直接指定routingKey
        amqpTemplate.convertAndSend("FirstKey", "[Direct,FirstKey] "+message);
        amqpTemplate.convertAndSend("SecondKey", "[Direct,SecondKey] "+message);

        // Exchange模式为topic，通过topic匹配关心该主题的队列
        amqpTemplate2.convertAndSend("msg.Third.send","[Topic,msg.Third.send] "+message);

        // 广播消息，与Exchange绑定的所有队列都会收到消息，routingKey为空
        amqpTemplate2.convertAndSend("MY_FANOUT_EXCHANGE",null,"[Fanout] "+message);
    }
}

public class FirstConsumer implements MessageListener {
    private Logger logger = LoggerFactory.getLogger(FirstConsumer.class);

    public void onMessage(Message message) {
        logger.info("The first consumer received message : " + message.getBody());
    }
}

public class SecondConsumer implements MessageListener {
    private Logger logger = LoggerFactory.getLogger(SecondConsumer.class);

    public void onMessage(Message message) {
        logger.info("The second consumer received message : " + message);
    }
}
....

springboot用法

消费者

@Configuration
@PropertySource("classpath:gupaomq.properties")
public class RabbitConfig {
    @Value("${com.gupaoedu.firstqueue}")
    private String firstQueue;

    @Value("${com.gupaoedu.secondqueue}")
    private String secondQueue;

    @Value("${com.gupaoedu.thirdqueue}")
    private String thirdQueue;

    @Value("${com.gupaoedu.fourthqueue}")
    private String fourthQueue;

    @Value("${com.gupaoedu.directexchange}")
    private String directExchange;

    @Value("${com.gupaoedu.topicexchange}")
    private String topicExchange;

    @Value("${com.gupaoedu.fanoutexchange}")
    private String fanoutExchange;

    // 创建四个队列
    @Bean("vipFirstQueue")
    public Queue getFirstQueue(){
        return new Queue(firstQueue);
    }

    @Bean("vipSecondQueue")
    public Queue getSecondQueue(){
        return new Queue(secondQueue);
    }

    @Bean("vipThirdQueue")
    public Queue getThirdQueue(){
        return  new Queue(thirdQueue);
    }

    @Bean("vipFourthQueue")
    public Queue getFourthQueue(){
        return  new Queue(fourthQueue);
    }

    // 创建三个交换机
    @Bean("vipDirectExchange")
    public DirectExchange getDirectExchange(){
        return new DirectExchange(directExchange);
    }

    @Bean("vipTopicExchange")
    public TopicExchange getTopicExchange(){
        return new TopicExchange(topicExchange);
    }

    @Bean("vipFanoutExchange")
    public FanoutExchange getFanoutExchange(){
        return new FanoutExchange(fanoutExchange);
    }

    // 定义四个绑定关系
    @Bean
    public Binding bindFirst(@Qualifier("vipFirstQueue") Queue queue, @Qualifier("vipDirectExchange") DirectExchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with("gupao.best");
    }

    @Bean
    public Binding bindSecond(@Qualifier("vipSecondQueue") Queue queue, @Qualifier("vipTopicExchange") TopicExchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with("*.gupao.*");
    }

    @Bean
    public Binding bindThird(@Qualifier("vipThirdQueue") Queue queue, @Qualifier("vipFanoutExchange") FanoutExchange exchange){
        return BindingBuilder.bind(queue).to(exchange);
    }

    @Bean
    public Binding bindFourth(@Qualifier("vipFourthQueue") Queue queue, @Qualifier("vipFanoutExchange") FanoutExchange exchange){
        return BindingBuilder.bind(queue).to(exchange);
    }

    /**
     * 在消费端转换JSON消息
     * 监听类都要加上containerFactory属性
     * @param connectionFactory
     * @return
     */
    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        factory.setAutoStartup(true);
        return factory;
    }
}

@Component
@PropertySource("classpath:gupaomq.properties")
@RabbitListener(queues = "${com.gupaoedu.firstqueue}", containerFactory="rabbitListenerContainerFactory")
public class FirstConsumer {

    @RabbitHandler
    public void process(@Payload Merchant merchant){
        System.out.println("First Queue received msg : " + merchant.getName());
    }

}

生产者

@Configuration
public class RabbitConfig {
    /**
     * 所有的消息发送都会转换成JSON格式发到交换机
     * @param connectionFactory
     * @return
     */
    @Bean
    public RabbitTemplate gupaoTemplate(final ConnectionFactory connectionFactory) {
        final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
        return rabbitTemplate;
    }
}

@Component
@PropertySource("classpath:gupaomq.properties")
public class RabbitSender {

    @Value("${com.gupaoedu.directexchange}")
    private String directExchange;

    @Value("${com.gupaoedu.topicexchange}")
    private String topicExchange;

    @Value("${com.gupaoedu.fanoutexchange}")
    private String fanoutExchange;

    @Value("${com.gupaoedu.directroutingkey}")
    private String directRoutingKey;

    @Value("${com.gupaoedu.topicroutingkey1}")
    private String topicRoutingKey1;

    @Value("${com.gupaoedu.topicroutingkey2}")
    private String topicRoutingKey2;

    @Autowired
    private MerchantService merchantService;


    // 自定义的模板，所有的消息都会转换成JSON发送
    @Autowired
    AmqpTemplate gupaoTemplate;

    public void send() throws JsonProcessingException {
        Merchant merchant =  new Merchant(1,"a direct msg : 中原镖局","汉中省解放路266号");
        gupaoTemplate.convertAndSend(directExchange,directRoutingKey, merchant);

        gupaoTemplate.convertAndSend(topicExchange,topicRoutingKey1, "a topic msg : shanghai.gupao.teacher");
        gupaoTemplate.convertAndSend(topicExchange,topicRoutingKey2, "a topic msg : changsha.gupao.student");

        // 发送JSON字符串
        ObjectMapper mapper = new ObjectMapper();
        String json = mapper.writeValueAsString(merchant);
        System.out.println(json);
        merchantService.update(merchant);
        gupaoTemplate.convertAndSend(fanoutExchange,"", json);
    }


}

service代码

@Override
    public int add(Merchant merchant) {
        int k = merchantMapper.add(merchant);
        System.out.println("aaa : "+merchant.getId());
        JSONObject title = new JSONObject();
        String jsonBody = JSONObject.toJSONString(merchant);
        title.put("type","add");
        title.put("desc","新增商户");
        title.put("content",jsonBody);
        gupaoTemplate.convertAndSend(topicExchange,topicRoutingKey, title.toJSONString());
        return k;
    }

消费端限流

//非自动确认消息的前提下，如果一定数目的消息（通过基于consume或者channel设置Qos的值）未被确认前，不进行消费新的消息。
channel.basicQos(2);
channel.basicConsume(QUEUE_NAME, false, consumer);

可靠性投递

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-eaepMLaG-1611231950956)(C:\Users\lichong\AppData\Roaming\Typora\typora-user-images\image-20201129223733782.png)]

1. ①代表消息从生产者发送到Broker

   RabbitMQ提供了两种服务端确认机制

   • Transaction（事务）模式

     try {
         channel.txSelect();
         channel.basicPublish("", QUEUE_NAME, null, (msg).getBytes());
         channel.txCommit();
         System.out.println("消息发送成功");
     } catch (Exception e) {
         channel.txRollback();
         System.out.println("消息已经回滚");
     }
     

     事务模式是阻塞的，一条消息没有发送完毕，不能发送下一条，太消耗服务器性能，不建议生产环境使用。

     SpringBoot中设置

     rabbitTemplate.setChannelTransacted(true);
     

   • Confirm（确认）模式

     /*单条确认*/
     // 开启发送方确认模式
     channel.confirmSelect();
     channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
     // 普通Confirm，发送一条，确认一条
     if (channel.waitForConfirms()) {
         System.out.println("消息发送成功" );
     }
     /*批量确认*/
     channel.confirmSelect();
     for (int i = 0; i < 5; i++) {
         // 发送消息
         // String exchange, String routingKey, BasicProperties props, byte[] body
         channel.basicPublish("", QUEUE_NAME, null, (msg +"-"+ i).getBytes());
     }
     // 批量确认结果，ACK如果是Multiple=True，代表ACK里面的Delivery-Tag之前的消息都被确认了
     // 比如5条消息可能只收到1个ACK，也可能收到2个（抓包才看得到）
     // 直到所有信息都发布，只要有一个未被Broker确认就会IOException
     channel.waitForConfirmsOrDie();
     System.out.println("消息发送完毕，批量确认成功");
     

     由于批量确认不能确定多少条确认一次，所以就有异步确认模式，一边发送一边确认。

     // 用来维护未确认消息的deliveryTag
     final SortedSet<Long> confirmSet = Collections.synchronizedSortedSet(new TreeSet<Long>());
     
     // 这里不会打印所有响应的ACK；ACK可能有多个，有可能一次确认多条，也有可能一次确认一条
     // 异步监听确认和未确认的消息
     // 如果要重复运行，先停掉之前的生产者，清空队列
     channel.addConfirmListener(new ConfirmListener() {
     public void handleNack(long deliveryTag, boolean multiple) throws IOException {
     System.out.println("Broker未确认消息，标识：" + deliveryTag);
     if (multiple) {
     // headSet表示后面参数之前的所有元素，全部删除
     confirmSet.headSet(deliveryTag + 1L).clear();
     } else {
     confirmSet.remove(deliveryTag);
     }
     // 这里添加重发的方法
     }
     public void handleAck(long deliveryTag, boolean multiple) throws IOException {
     // 如果true表示批量执行了deliveryTag这个值以前（小于deliveryTag的）的所有消息，如果为false的话表示单条确认
     System.out.println(String.format("Broker已确认消息，标识：%d，多个消息：%b", deliveryTag, multiple));
     if (multiple) {
     // headSet表示后面参数之前的所有元素，全部删除
     confirmSet.headSet(deliveryTag + 1L).clear();
     } else {
     // 只移除一个元素
     confirmSet.remove(deliveryTag);
     }
     System.out.println("未确认的消息:"+confirmSet);
     }
     });
     
     // 开启发送方确认模式
     channel.confirmSelect();
     for (int i = 0; i < 10; i++) {
     long nextSeqNo = channel.getNextPublishSeqNo();
     // 发送消息
     // String exchange, String routingKey, BasicProperties props, byte[] body
     channel.basicPublish("", QUEUE_NAME, null, (msg +"-"+ i).getBytes());
     confirmSet.add(nextSeqNo);
     }
     System.out.println("所有消息:"+confirmSet);
     

     springboot中confirm模式实在Channel上开启的，RabbitTemplate对Channel进行了封装

     @Bean
     public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
         RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
         rabbitTemplate.setMandatory(true);
         rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback(){
             public void returnedMessage(Message message,
             int replyCode,
             String replyText,
             String exchange,
             String routingKey){
             System.out.println("回发的消息：");
             System.out.println("replyCode: "+replyCode);
             System.out.println("replyText: "+replyText);
             System.out.println("exchange: "+exchange);
             System.out.println("routingKey: "+routingKey);
             }
     	});
     
         rabbitTemplate.setChannelTransacted(true);
     
         rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
             public void confirm(CorrelationData correlationData, boolean ack, String cause) {
             if (!ack) {
                 System.out.println("发送消息失败：" + cause);
                 throw new RuntimeException("发送异常：" + cause);
             }
         }
     });
     return rabbitTemplate;
     }
     

2. ②代表消息从Exchange路由到Queue

   这个环节出错有两种可能，一是routingkey错误，二是队列不存在。（生产环境基本不会出现这两种情况）

   channel.addReturnListener(new ReturnListener() {
       public void handleReturn(int replyCode,
                                String replyText,
                                String exchange,
                                String routingKey,
                                AMQP.BasicProperties properties,
                                byte[] body)
           throws IOException {
           System.out.println("=========监听器收到了无法路由，被返回的消息============");
           System.out.println("replyText:"+replyText);
           System.out.println("exchange:"+exchange);
           System.out.println("routingKey:"+routingKey);
           System.out.println("message:"+new String(body));
       }
   });
   

   Springboot消息回发的方式

   // 消息是否必须路由到一个队列中，配合Return使用
   rabbitTemplate.setMandatory(true);
   // 为RabbitTemplate设置ReturnCallback
   rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
   	@Override
       public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                   try {
                       System.out.println("--------收到无法路由回发的消息--------");
                       System.out.println("replyCode:" + replyCode);
                       System.out.println("replyText:" + replyText);
                       System.out.println("exchange:" + exchange);
                       System.out.println("routingKey:" + routingKey);
                       System.out.println("properties:" + message.getMessageProperties());
                       System.out.println("body:" + new String(message.getBody(), "UTF-8"));
                   } catch (UnsupportedEncodingException e) {
                       e.printStackTrace();
                   }
               }
    });
   

   创建交换机的时候，从属性中指定备份交换机

   // 在声明交换机的时候指定备份交换机
   Map<String,Object> arguments = new HashMap<String,Object>();
   arguments.put("alternate-exchange","ALTERNATE_EXCHANGE");
   channel.exchangeDeclare("TEST_EXCHANGE","topic", false, false, false, arguments);
   
   // 发送到了默认的交换机上，由于没有任何队列使用这个关键字跟交换机绑定，所以会被退回
   // 第三个参数是设置的mandatory，如果mandatory是false，消息也会被直接丢弃
   channel.basicPublish("TEST_EXCHANGE","qingshan2673",true, properties,"只为更好的你".getBytes());
   

3. ③代表消息在Queue中存储

   如果没有消费者，队列一直存在在数据库中。如果服务器或硬件发生故障，比如宕机、重启、关闭。所以要把消息本身和元数据都保存到磁盘。

   • 队列持久化

     // 声明队列
     // String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
     //durable:没有持久化的队列，保存在内存中，服务重启后队列和消息都会丢失。
     //exclusive:排他性队列的特点：只对首次声明他的连接可见。会在连接断开的时候自动删除。
     //autoDelete:没有消费者连接的时候自动删除。
     channel.queueDeclare(QUEUE_NAME, false, false, false, null);
     

   • 交换机持久化

     @Bean("gpexchange")
     public DirectExchange exchange() {
     	return new DirectExchange("GP_RELIABLE_RECEIVE_EXCHANGE", true, false, new HashMap<>());
     }
     

   • 消息持久化

     AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
                     .deliveryMode(2)   // 2代表持久化
                     .contentEncoding("UTF-8")  // 编码
                     .expiration("10000")  // TTL，过期时间
                     .headers(headers) // 自定义属性
                     .priority(5) // 优先级，默认为5，配合队列的 x-max-priority 属性使用
                     .messageId(String.valueOf(UUID.randomUUID()))
                     .build();
     

     如果消息没有持久化，保存在内存中，队列还在，但是消息在重启后会消失。

     如果只有一个RabbitMQ节点，即使交换机、队列、消息做了持久化，如果服务崩溃或者硬件发生故障，RabbitMQ的服务一样是不可用的。所以为了提高MQ服务的可用性，保障消息的传输，我们需要多个RabbitMQ的节点，也就是集群。

4. ④代表消费者订阅Queue并消费消息

   RabbitMQ提供了消息的确认机制，消费者可以自动或手动发送ACK给服务端。

   没有收到ACK的消息，消费者断开连接后，RabbitMQ会把这条消息发送给其他消费者。如果没有其他消费者，消费者重启后会重新消费这条消息，重复执行业务逻辑。

   消费者有两种方式给broker应答。一种是自动ACK，一种是手动ACK。

   自动ACK，也是默认的情况。也就是没有在消费者处编写ACK的代码，消费者会在***收到消息的时候***就自动发送ACK，并不关心有没有正常消费。

   如果想要等消息执行完成后才发送ACK，需要先把自动ACK设置成手动ACK。把autoAck设置成false。

   // String queue, boolean autoAck, Consumer callback
   channel.basicConsume(QUEUE_NAME, false, consumer);
   

   Springboot中设置

   spring.rabbitmq.listener.direct.acknowledge-mode=manual
   spring.rabbitmq.listener.simple.acknowledge-mode=manual
   

   代码中设置

   // 设置消息确认模式为手动模式
   messageListenerContainer.setAcknowledgeMode(AcknowledgeMode.MANUAL);
   

   None:自动ACK

   MANUAL：手动ACK

   AUTO:如果方法未抛出异常，则发送ack。如果方法抛出异常，并且不是AmqpRejectAndDontRequeueException则发送nack，并且重新进入队列。如果抛出异常，并且是AmqpRejectAndDontRequeueException异常则发送nack不会重新进入队列。

   消费者调用Ack:

   @RabbitHandler
   public void process(String msgContent,Channel channel, Message message) throws IOException {
   	System.out.println("Second Queue received msg : " + msgContent );
   	channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }
   

   如果消费出了问题，也有拒绝消息的指令，而且还可以让消息重新入队给其他消费者消费。

   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("Received message : '" + msg + "'");
           if (msg.contains("拒收")){
               // 拒绝消息
               // requeue：是否重新入队列，true：是；false：直接丢弃，相当于告诉队列可以直接删除掉
               // TODO 如果只有这一个消费者，requeue 为true 的时候会造成消息重复消费
               channel.basicReject(envelope.getDeliveryTag(), false);
           } else if (msg.contains("异常")){
               // 批量拒绝
               // requeue：是否重新入队列
               // TODO 如果只有这一个消费者，requeue 为true 的时候会造成消息重复消费
               channel.basicNack(envelope.getDeliveryTag(), true, false);
           } else {
               // 手工应答
               // 如果不应答，队列中的消息会一直存在，重新连接的时候会重复消费
               channel.basicAck(envelope.getDeliveryTag(), true);
           }
       }
   };
   

   生产者确定消费者是否消费成功的两种方式：

   1. 消费者收到消息，处理完毕后，调用生产者的API（是否破化解耦）
   2. 消费者收到消息，处理完毕后，发送一条响应消息给生产者。

补偿机制

如果生产者的API没有别调用，也没有收到消费者的响应消息，生产者和消费者应该约定一个超时时间，对于超过这个时间没有得到响应的消息，才确定为消费失败。

消费失败后,需要重新发送消息。

1. 谁来重发？

   可以创建一个数据库表记录消费失败的消息，由定时任务定时重新发送。（会消耗性能，消耗数据库存储空间）

2. 隔多久发送一次？由定时任务的时间决定。

3. 一共重发多少次？可以设置为几次，在消息表里记录次数实现重新发送。

4. 重发什么内容？发送一模一样的消息

消息的幂等性

如果消费者状态时正常的，每一条消息都可以正常处理，只是在响应或者调用API的时候出现问题。为了避免相同消息的重复处理，必须采取一定的措施。RabbitMQ服务端是没有这种控制的，只能在消费端处理。

重复消息的两个原因

1. 生产者的问题，环节①重复发送消息，比如在开启了Confirm模式但未收到确认，消费者重复投递。
2. 环节④出了问题，由于消费者未发送ACK或者其他原因，消息重复消费。

对于重复发送的消息，可以对每一条消息生成一个唯一的业务ID，通过日志或者消息落库来做重复控制。

最终一致

约定一个标准，获取核心系统的日志文件，解析，登记成数据，然后跟自己记录的流水比较，跟核心系统保持一致。

消息的顺序性

消息的顺序性是指消费者消费的顺序跟生产者生产的消息顺序是一致的。一个队列有多个消费者时无法保证顺序。

只有一个队列有一个消费者时才能保证顺序消费。

集群和高可用

集群目的：实现高可用和负载均衡。

如何支持集群？

由于是Erlang开发，所以天然支持集群。

通过.erlang.cookie(默认路径：/var/lib/rabbitmq)来验证身份，需要在所有节点上保持一致。

RabbitMQ节点类型

集群有两种节点类型：一种是磁盘节点（disc node），一种是内存节点(ram node)

磁盘节点：将元数据（队列名字，属性，交换机的名字、属性，绑定，vhost）放在磁盘中。默认磁盘节点。

内存节点：将元数据放在内存中。

集群中至少需要一个磁盘节点来持久化元数据，否则全部内存节点崩溃时，就无从同步元数据。

集群的配置步骤：

1. 配置hosts以便互相通信
2. 同步erlang.cookie
3. 加入集群（join cluster 命令）

rabbitmq有两种集群模式：普通集群模式和镜像队列集群模式。

普通集群

不同的节点之间只会相互同步元数据（交换机、队列、绑定关系，vhost的定义），而不会同步消息。如果要保证队列的高可用，不能用这种集群。

镜像集群

消息内容会在镜像节点同步，可用性更高。但是会降低系统性能。

集群模式可以通过UI或者CLI或者HTTP操作。

高可用

通过VRRP协议，这个组件就是Keepalived,它具有Load Balane和high Availability的功能。

生产环境运维监控可以用zabbix+grafana实现，主要关注：磁盘，内存，连接数。

发送消息：可以用json数组格式发送，建议单条消息不要超过4M（4096kb）

kafka

kafka的服务叫做broker,默认是9092端口，生产者和消费者都是跟一个broker建立连接才能实现消息的收发。

生产者对应的封装类是ProducerRecord,消费者对应的封装类是ConsumerRecord,消息在传输过程中需要***。

生产者

生产者不是逐条发送消息给broker，而是批量发送的。多少条发送一次由一个参数决定。

proc.put("batch.size",16384);

消费者

kafka的消费是pull模式。而且可以控制一次到底取多少条消息,默认是500，可以在poll方法里面指定。

max.poll.records

因为在push模式下，如果消息的生产速度远远大于消费者消费的速度，那消费者就会不堪重负，直接挂掉。

Topic

在kakfa中，队列叫topic，是一个逻辑的概念，可以理解为消息的组合。生产者和topic以及消费者和topic都是多对多的关系。

生产者发送消息时，如果topic不存在，会自动创建，由一个参数控制。

默认为true。

auto.create.topics.enable

Partition和Cluster

如果一个topic中的消息过多的话，会带来两个问题：

第一个不方便横向扩展

第二个是并发或者负载的问题。所有的客户端操作都在一个topic上，在高并发的场景下性能会大大降低。

为了解决上面的问题，引入了分区，就是把topic拆分。

分区在创建topic的时候指定，每个topic至少有一个分区

./kafka-topics.sh --create --zookper localhost:2181 --replication-factor 1 --partitions 1 --topic gptest

如果没有指定分区，默认的分区是一个，这个参数可以修改。

num.partitions=1

partitions是分区数，replication-factor是分区副本数。

partition分区里面的消息读取后不会被删除。

Partition副本Replica机制

如果partition的数据只存储一份，在发生网络或者硬件故障的时候，该分区的数据就无法访问或者无法恢复了。

kafka在0.8版本之后增加了副本机制。

每个partition可以有若干个副本（replica），副本必须在不同的broker上面。

服务端有一个参数控制默认的副本数

offsets.topic.replication.factor

Segment

如果一个partition只有一个log文件，消息不断地追加，这个log我呢见也会变得越来越大，这个时候要检索数据库效率就会很低。所以干脆把partition再做一个切分，切分出来的单位就叫段。实际上kafka的存储文件是划分成段来存储的。

默认存储路径：/tmp/kafka-logs/

每个segment都至少有1个数据文件和2个索引文件，这三个文件是成套出现的。

一个segment默认大小是1073741824bytes（1G），由下面参数控制

log.segment.bytes

Consumer Group

如果生产者消息的速率过快，会造成消息在broker的堆积，影响broker的性能。通过增加消费者的数量提升消费速率，消费者过多怎么确定消费的在同一个topic？

所以引入了一个Consumer Group消费组的概念。在代码中通过group id来配置。消费同一个topic的消费者不一定是同一个组，只有group id相同的消费者才是一个消费者组。

Consumer Offset

partition里面的消息是顺序写入的，被读取之后为了不被删除。所以对消息进行编号，用来标识一条唯一的消息。这个编号叫做offset，偏移量。offset记录着下一条将要发送给consumer的消息序号。偏移量保存在服务端。

java开发

public class SimpleConsumer {
    public static void main(String[] args) {
        Properties props= new Properties();
        //props.put("bootstrap.servers","192.168.44.161:9093,192.168.44.161:9094,192.168.44.161:9095");
        props.put("bootstrap.servers","192.168.242.110:9092");
        props.put("group.id","gp-test-group");
        // 是否自动提交偏移量，只有commit之后才更新消费组的 offset
        props.put("enable.auto.commit","true");
        // 消费者自动提交的间隔
        props.put("auto.commit.interval.ms","1000");
        // 从最早的数据开始消费 earliest | latest | none
        props.put("auto.offset.reset","earliest");
        props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String,String> consumer=new KafkaConsumer<String, String>(props);
        // 订阅topic
        consumer.subscribe(Arrays.asList("mytopic"));

        try {
            while (true){
                ConsumerRecords<String,String> records=consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String,String> record:records){
                    System.out.printf("offset = %d ,key =%s, value= %s, partition= %s%n" ,record.offset(),record.key(),record.value(),record.partition());
                }
            }
        }finally {
            consumer.close();
        }
    }
}

public class  SimpleProducer {
    public static void main(String[] args) {
        Properties pros=new Properties();
        //pros.put("bootstrap.servers","192.168.44.161:9093,192.168.44.161:9094,192.168.44.161:9095");
        pros.put("bootstrap.servers","192.168.242.110:9092");
        pros.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
        pros.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
        // 0 发出去就确认 | 1 leader 落盘就确认| all(-1) 所有Follower同步完才确认
        pros.put("acks","1");
        // 异常自动重试次数
        pros.put("retries",3);
        // 多少条数据发送一次，默认16K
        pros.put("batch.size",16384);
        // 批量发送的等待时间
        pros.put("linger.ms",5);
        // 客户端缓冲区大小，默认32M，满了也会触发消息发送
        pros.put("buffer.memory",33554432);
        // 获取元数据时生产者的阻塞时间，超时后抛出异常
        pros.put("max.block.ms",3000);

        // 创建Sender线程
        Producer<String,String> producer = new KafkaProducer<String,String>(pros);

        for (int i =0 ;i<10;i++) {
            producer.send(new ProducerRecord<String,String>("mytopic",Integer.toString(i),Integer.toString(i)));
            // System.out.println("发送:"+i);
        }

        //producer.send(new ProducerRecord<String,String>("mytopic","1","1"));
        //producer.send(new ProducerRecord<String,String>("mytopic","2","2"));

        producer.close();
    }
}

kafka与springboot实战

server.port=7271
#spring.kafka.bootstrap-servers=192.168.44.161:9093,192.168.44.161:9094,192.168.44.161:9095
spring.kafka.bootstrap-servers=192.168.44.160:9092

# producer
spring.kafka.producer.retries=1
spring.kafka.producer.batch-size=16384
spring.kafka.producer.buffer-memory=33554432
spring.kafka.producer.acks=all
spring.kafka.producer.properties.linger.ms=5
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer

# consumer
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.enable-auto-commit=true
spring.kafka.consumer.auto-commit-interval=1000
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

消费者使用@KafkaListener监听topic

@Component
public class ConsumerListener {
    @KafkaListener(topics = "springboottopic",groupId = "springboottopic-group")
    public void onMessage(String msg){
        System.out.println("----收到消息："+msg+"----");
    }
}

生产者

@Component
public class KafkaProducer {
    @Autowired
    private KafkaTemplate<String,Object> kafkaTemplate;

    public String send(@RequestParam String msg){
        kafkaTemplate.send("springboottopic", msg);
        return "ok";
    }
}

测试先启动Application,在运行单元测试

@SpringBootTest
class KafkaTests {

    @Autowired
    KafkaProducer producer;

    // 消费者：先启动 kafkaApp
    @Test
    void testSendMsg() {
        long time = System.currentTimeMillis();
        System.out.println("----"+time +"，已经发出----");
        producer.send("qingshan penyuyan, " +time);
    }
}

kafka继承canal实现数据同步

canal利用mysql的binlog日志功能，把自己伪装成一个slave节点，不断请求最新的binlog。canal会解析binlog的内容，把内容发给关注数据库变动的接收者，完成后续逻辑的处理。

Canal是一个纯java开发的数据同步工具，可以支持binlog增量订阅的功能。binlog设置成row模式以后，不仅能获取到执行的每一个增删改的脚本，同时能获取到修改前和修改后的数据。

工作流程：数据变动-产生binlog信息-canal服务获取binlog信息-发送MQ消息-消费者消费MQ消息，完成后续逻辑处理。

进阶功能

消息幂等性

如果消费失败了，消息需要重发。但是不清楚消费者是不是真的消费失败的情况下，有可能出现消息重复的情况。

消息重复需要在消费端解决，也就是在消费者实现幂等性。

考虑到所有的消费者都要做一样的实现，kafka干脆在broker实现了消息的重复性判断，大大的解放了消费者的双手。

去重肯定要依赖于生产者的消息唯一的标识，不然是没有办法知道是不是同一条消息的。通过配置：

props.put("enable.idempotence",true);

enable.idempotence设置成true之后，producer自动升级成幂等性的producer，kafka会自动去重。有两个实现机制：

1. PID（Producer ID）,幂等性的生产者每个客户端都有一个唯一的编号。

2. sequence number，幂等性的生产者发送的每条消息都带有相应的sequence number，server端就是根据这个值来判断数据是否重复。如果说sequence number比服务端已经记录的值要小，那肯定是出现消息重复了。

   1. 由于sequence number并不是全局有序的，不能保证所有时间上的幂等性。所以他的作用范围是有限的
   2. 只能保证单分区上的幂等性，即一个幂等性的producer能够保证某个主题的一个分区上不出现重复消息。
   3. 只能实现单会话的幂等性，这里的会话指的是producer进程的一次运行。当重启了producer进程之后，幂等性不保证。

   如果要实现多个分区的消息的原子性，就要用到kafka的事务机制了。

生产者事务

生产者事务是kakfa2017引入的新特性，通过事务，kafka可以保证生产者会话的消息幂等发送。

public class TransactionProducer {
    public static void main(String[] args) {
        Properties props=new Properties();
        //props.put("bootstrap.servers","192.168.44.161:9093,192.168.44.161:9094,192.168.44.161:9095");
        props.put("bootstrap.servers","192.168.44.160:9092");
        props.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
        // 0 发出去就确认 | 1 leader 落盘就确认| all或-1 所有Follower同步完才确认
        props.put("acks","all");
        // 异常自动重试次数
        props.put("retries",3);
        // 多少条数据发送一次，默认16K
        props.put("batch.size",16384);
        // 批量发送的等待时间
        props.put("linger.ms",5);
        // 客户端缓冲区大小，默认32M，满了也会触发消息发送
        props.put("buffer.memory",33554432);
        // 获取元数据时生产者的阻塞时间，超时后抛出异常
        props.put("max.block.ms",3000);

        props.put("enable.idempotence",true);
        // 事务ID，唯一
        props.put("transactional.id", UUID.randomUUID().toString());

        Producer<String,String> producer = new KafkaProducer<String,String>(props);

        // 初始化事务
        producer.initTransactions();

        try {
            producer.beginTransaction();
            producer.send(new ProducerRecord<String,String>("transaction-test","1","1"));
            producer.send(new ProducerRecord<String,String>("transaction-test","2","2"));
            // Integer i = 1/0;
            producer.send(new ProducerRecord<String,String>("transaction-test","3","3"));
            // 提交事务
            producer.commitTransaction();
        } catch (KafkaException e) {
            // 中止事务
            producer.abortTransaction();
        }


        producer.close();
    }
}

特性

• 高吞吐、低延迟：最大特点就是收发消息快，每秒可以处理几十万条消息，他的延迟只有几毫秒
• 高伸缩性：通过增加分区partition来实现扩容。不同的分区可以在不同的broker中。通过zk来管理broker实现扩展，zk管理consumer可以实现负载。
• 持久性、可靠性：kafka能够允许数据的持久化存储，消息被持久化到磁盘，支持数据备份防止数据丢失。
• 容错性：允许集群中的节点失败，某个节点宕机，kafka集群能正常工作。
• 高并发：支持数千个客户端同时读写。

	产品侧重	性能	消息顺序	消息路由和分发	延迟消息。死信队列	消息的留存
RabbitMq	消息代理	主要是push		更加灵活	支持	不留存
Kafka	流式消息处理、消息引擎	只有pull，吞吐量更高	分区消息有序，能保证消息的顺序		不支持	消费完会清除
RocketMq		只有pull

RocketMQ

RocketMQ常用管理命令

https://blog.csdn.net/gwd1154978352/article/details/80829534

RocketMQ默认配置

https://www.cnblogs.com/jice/p/11981107.html

broker

RocketMQ的服务叫做broker，broker的作用是存储和转发消息。单机大约能承受10万QPS。

为了提高可靠性，每个Broker可以有自己的副本。默认情况下，读写都发生在master上。在slaveReadEnable=true的情况下，slave也可以参与读负载。但是默认只有BrokerId=1的slave才会参与读负载，而且是在master消费慢的情况下，由whichBrokerWhenConsumeSlowly这个参数决定。

topic

topic用于将消息按主题做划分。跟kafka不同的是，在RocketMQ中，Topic是一个逻辑概念，消息不是按Topic划分存储的。Producer将消息发往指定的topic，consumer订阅这个topic就可以收到相应的消息。跟kafka一样，如果topic不存在，会自动创建，BrokerConfig：

private boolean autoCreateTopicEnable = true;

topic跟生产者和消费者是多对多的关系。

NameServer

可以把NameServer理解为是RocketMQ的路由中心，每一个NameServer节点都保存着全量的路由信息，为了保证高可用，NameServer也可以做集群的部署。Broker会在NameServer上注册自己，Producer和Consumer用NameServer来发现Broker。

NameServer作为路由中心怎么工作？

每个Broker节点在启动时，都会根据配置遍历NameServer列表。Broker与每个NameServer建立TCP长连接，注册自己的信息，之后每隔30s发送心跳信息。除了主从注册，还有定时探活。每个NameServer每隔10s检查一下各个Broker的最近一次心跳时间，如果发现某个Broker超过120s都没发送心跳，就认为这个Broker已经挂掉 了，会将其从路有消息里移除。

一致性问题

1. 服务注册

   因为没有master，Broker每隔30s会向所有的NameServer发送心跳信息。

2. 服务剔除

   • 如果broker正常关闭，连接就断开了，netty的通道关闭监听器会监听到连接断开事件，然后将这个broker剔除掉
   • 如果broker异常关闭，NamServer的定时任务每10s扫描Broker列表，如果某个broker的心跳包的最新时间戳超过当前时间120s，就会被剔除。

3. 路由发现

   生产者：发送消息的时候，根据topic从NameServer获取路由信息。

   消费者：一般是订阅固定的topic。在MQClientInstance类中由定时任务定期更新NamServer信息，默认30s获取一次。

producer

生产者，会定时从NameServer拉取路由信息，然后根据路由信息与指定的Broker建立TCP长连接，从而将消息发送到Broker中。发送逻辑一致的Producer可以组成一个Group。Producer写数据只能操作master节点。

consumer

消费者有两种消费方式：一种是集群消费，一种是广播消费。

Message Queue

作用跟kafka里面的partition分片类似。

在创建topic的时候会指定队列的数量，一个叫writeQueueNums写队列的数量，一个readQueueNums读队列的数量。写队列的数量决定了有几个Message Queue，读队列的数量决定了有几个线程来消费这些Message Queue。

java开发

生产者：

public class Producer {
    public static void main(String[] args) throws MQClientException, InterruptedException {
        DefaultMQProducer producer = new DefaultMQProducer("my_test_producer_group");
        producer.setNamesrvAddr("192.168.44.163:9876;192.168.44.164:9876");
        producer.start();
        for (int i = 0; i < 6; i++){
            try {
                // tags 用于过滤消息 keys 索引键,多个用空格隔开,RocketMQ可以根据这些key快速检索到消息
                Message msg = new Message("q-2-1",
                        "TagA",
                        "2673",
                        ("RocketMQ "+String.format("%05d", i)).getBytes());

                SendResult sendResult = producer.send(msg);
                System.out.println(String.format("%05d", i)+" : "+sendResult);
            }
            catch (Exception e) {
                e.printStackTrace();
            }
        }
        producer.shutdown();
    }
}

消费者：

public class SimpleConsumer {
    public static void main(String[] args) throws MQClientException {
        // Instantiate with specified consumer group name.
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("my_test_consumer_group");

        // Specify name server addresses.
        consumer.setNamesrvAddr("192.168.44.163:9876;192.168.44.164:9876");
        consumer.setMessageModel(MessageModel.BROADCASTING);

        // Subscribe one more more topics to consume.
        consumer.subscribe("q-2-1", "*");
        // Register callback to execute on arrival of messages fetched from brokers.
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                           ConsumeConcurrentlyContext context) {
                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                for(MessageExt msg : msgs){
                    String topic = msg.getTopic();
                    String messageBody="";
                    try {
                        messageBody = new String(msg.getBody(),"utf-8");
                    } catch (UnsupportedEncodingException e) {
                        e.printStackTrace();
                        // 重新消费
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                    String tags = msg.getTags();
                    System.out.println("topic:"+topic+",tags:"+tags+",msg:"+messageBody);
                }

                // 消费成功
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //Launch the consumer instance.
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }
}

springboot集成

消费者

@Component
@RocketMQMessageListener(topic = "springboot-topic",consumerGroup = "qs-consumer-group",
        //selectorExpression = "tag1",selectorType = SelectorType.TAG,
        messageModel = MessageModel.CLUSTERING, consumeMode = ConsumeMode.CONCURRENTLY)
public class MessageConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        try {
            System.out.println("----------接收到rocketmq消息:" + message);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

生成者

@Component
public class MessageSender {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    public void syncSend(){
        /**
         * 发送可靠同步消息 ,可以拿到SendResult 返回数据
         * 同步发送是指消息发送出去后，会在收到mq发出响应之后才会发送下一个数据包的通讯方式。
         * 这种方式应用场景非常广泛，例如重要的右键通知、报名短信通知、营销短信等。
         *
         * 参数1： topic:tag
         * 参数2:  消息体 可以为一个对象
         * 参数3： 超时时间 毫秒
         */
        SendResult result= rocketMQTemplate.syncSend("springboot-topic:tag","这是一条同步消息",10000);
        System.out.println(result);
    }

    /**
     * 发送 可靠异步消息
     * 发送消息后，不等mq响应，接着发送下一个数据包。发送方通过设置回调接口接收服务器的响应，并可对响应结果进行处理。
     * 异步发送一般用于链路耗时较长，对于RT响应较为敏感的业务场景，例如用户上传视频后通过启动转码服务，转码完成后通推送转码结果。
     *
     * 参数1： topic:tag
     * 参数2:  消息体 可以为一个对象
     * 参数3： 回调对象
     */
    public void asyncSend() throws Exception{

        rocketMQTemplate.asyncSend("springboot-topic:tag1", "这是一条异步消息", new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("回调sendResult:"+sendResult);
            }
            @Override
            public void onException(Throwable e) {
                System.out.println(e.getMessage());
            }
        });
        TimeUnit.SECONDS.sleep(100000);
    }

    /**
     * 发送单向消息
     * 参数1： topic:tag
     * 参数2:  消息体 可以为一个对象
     */
    public void sendOneWay(){
        rocketMQTemplate.sendOneWay("springboot-topic:tag1", "这是一条单向消息");
    }

    /**
     * 发送单向的顺序消息
     */
    public void sendOneWayOrderly(){
        for(int i=0;i<10;i++){
            rocketMQTemplate.sendOneWayOrderly("springboot-topic:tag1", "这是一条顺序消息"+i,"2673");
            rocketMQTemplate.sendOneWayOrderly("springboot-topic:tag1", "这是一条顺序消息"+i,"2673");
        }
    }