RocketMQ源码--2--RocketMQ的简单使用

「爱情、让人受尽委屈。」 2022-10-16 09:54 200阅读 0赞

# `RocketMQ`源码–`2`–`RocketMQ`的简单使用 #

### 文章目录 ###

*  \`RocketMQ\`源码--\`2\`--\`RocketMQ\`的简单使用
 *  前言：
 *  1 依赖
 *  2 发送消息
 *   *  2.1 同步发送
     *  2.1 异步发送
     *  2.3 单向发送
 *  3 消费消息
 *  4 顺序消息
 *   *  4.1 顺序消息生产
     *  4.2 顺序消息消费
 *  5 延时消息

# 前言： #

> 使用`rocketmq`其实非常简单，官方已经给出了各种情况下的简单示例，下面我们就跟随官方的案例来看一下它是如何使用的
> 
> [官方教程地址][Link 1]

# 1 依赖 #

`pom`中引入这个依赖即可

<dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-client</artifactId>
        <version>4.3.0</version>
    </dependency>

# 2 发送消息 #

> `RocketMQ`支持好几种消息发送方式，都有各自的应用场景
> 
> 1.  同步发送
> 2.  异步发送
> 3.  单向发送

## 2.1 同步发送 ##

同步发送的意思就是说：生产者必须收到`rocketmq`对该条消息的回复才会继续往下运行，否则就一直阻塞在该行代码那里。

这种可靠性同步地发送方式使用的比较广泛，比如：重要的消息通知，短信通知。

public class SyncProducer { 
        public static void main(String[] args) { 
            // rocketmq提供的一个发消息的生产者
            DefaultMQProducer defaultMQProducer = new DefaultMQProducer("sync_test");
            // 必须设置NameServer的地址
            defaultMQProducer.setNamesrvAddr("localhost:9876");
            try { 
                // 启动Producer实例
                defaultMQProducer.start();
                for (int i = 0; i < 10; i++) { 
                    Message message = new Message("test_topic", "tagA", ("第" + i + "条消息").getBytes(RemotingHelper.DEFAULT_CHARSET));
                    // 必须要收到回复，否则会一直阻塞
                    SendResult sendResult = defaultMQProducer.send(message);
                    System.out.printf("%s%n", sendResult);
                }
            } catch (Exception e) { 
                e.printStackTrace();
            }
        }
    }

> `NameServer`就相当于一个注册中心，所有的`broker`（真正的`mq`）都将自己的注册到`NameServer`上，生产者发送消息的时候，先从`NameServer`中获取一个`broker`的地址，然后才将消息发送到这个`broker`上。

## 2.1 异步发送 ##

异步发送和同步发送唯一区别就是：同步发送要等发送结果，而异步则不用。

异步消息通常用在对响应时间敏感的业务场景，即发送端不能容忍长时间地等待`Broker`的响应。

public class AsyncProducer { 
    
        public static void main(String[] args) { 
            // rocketmq提供的一个发消息的生产者
            DefaultMQProducer defaultMQProducer = new DefaultMQProducer("async_test");
            // 必须设置NameServer的地址
            defaultMQProducer.setNamesrvAddr("localhost:9876");
            try { 
                // 启动Producer实例
                defaultMQProducer.start();
                for (int i = 0; i < 10; i++) { 
                    Message message = new Message("test_topic", "tagA", ("第" + i + "条消息").getBytes(RemotingHelper.DEFAULT_CHARSET));
                    // 发完就不管了，直接运行下一行代码
                    defaultMQProducer.send(message, new SendCallback() { 
                        /** * 收到mq回复发送成功后，回调此方法 * @param sendResult */
                        @Override
                        public void onSuccess(SendResult sendResult) { 
                            System.out.printf("%s%n", sendResult);
                        }
    
                        /** * 收到mq回复发送失败后，回调此方法 * @param e */
                        @Override
                        public void onException(Throwable e) { 
                            System.out.printf("%s%n", e);
                        }
                    });
    
                }
            } catch (Exception e) { 
                e.printStackTrace();
            }
        }
    }

## 2.3 单向发送 ##

这种方式主要用在不特别关心发送结果的场景，例如日志发送。

public class OnewayProducer { 
        public static void main(String[] args) { 
            // rocketmq提供的一个发消息的生产者
            DefaultMQProducer defaultMQProducer = new DefaultMQProducer("oneway_test");
            // 必须设置NameServer的地址
            defaultMQProducer.setNamesrvAddr("localhost:9876");
            try { 
                // 启动Producer实例
                defaultMQProducer.start();
                for (int i = 0; i < 10; i++) { 
                    Message message = new Message("test_topic", "tagA", ("第" + i + "条消息").getBytes(RemotingHelper.DEFAULT_CHARSET));
                    // 发完就不管了，不关心结果，直接运行下一行代码
                    defaultMQProducer.sendOneway(message);
                }
            } catch (Exception e) { 
                e.printStackTrace();
            }
        }
    }

# 3 消费消息 #

> 下面是一个简单的消费者示例，重点地方都已经写好注释

public class Consumer { 
        public static void main(String[] args) { 
    
            // 实例化一个消费者
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_test");
            // 设置NameServer的地址
            consumer.setNamesrvAddr("localhost:9876");
            try { 
                // 设置消费者订阅的主题topic和tag
                consumer.subscribe("test_topic","tagA");
                // 注册一个监听器来消费消息
                consumer.registerMessageListener(new MessageListenerConcurrently() { 
                    /** * It is not recommend to throw exception,rather than returning ConsumeConcurrentlyStatus.RECONSUME_LATER if * consumption failure * * @param msgs msgs.size() >= 1<br> DefaultMQPushConsumer.consumeMessageBatchMaxSize=1,you can modify here * @param context * @return The consume status */
                    @Override
                    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { 
                        ListIterator<MessageExt> iterator = msgs.listIterator();
                        while (iterator.hasNext()){ 
                            MessageExt messageExt = iterator.next();
                            try { 
                                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), new String(messageExt.getBody(),"utf-8"));
                            } catch (UnsupportedEncodingException e) { 
                                e.printStackTrace();
                            }
                        }
                        // 向mq返回一个消费成功的标识
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                });
                // 启动消费者
                consumer.start();
                System.out.printf("Consumer Started.%n");
            } catch (MQClientException e) { 
                e.printStackTrace();
            }
        }
    }

# 4 顺序消息 #

> 什么是顺序消息：就是按照发送消息的顺序来进行消费。比如说，生产者先后发送了`3`条消息`A`、`B`、`C`，那么消费者消费的时候也需要按照这个顺序先消费`A`，再消费`B`，最后消费`C`。那么如何做到呢？
> 
> 首先你得明白`rocketmq`收发消息的原理：
> 
> 其实`broker`收到消息后，会将消息分配到它下面的`MessageQueue`中，`broker`与`MessageQueue`的关系是一对多，即一个`broker`下面可能会存在多个`MessageQueue`。默认情况下，一个`broker`收到的消息会被均匀的存储到它下面的`MessageQueue`上，`MessageQueue`就是我们常说的消息队列，`rocketmq`收到的所有消息都存到这些消息队列中。
> 
> 消息存储到`MessageQueue`上后，消费者又是如何消费的呢？
> 
> 和生产者一样，消费这也是先从`NameServer`中获取一个`broker`的地址，连上`broker`后，均匀的从它下面的`MessageQueue`中获取消息，消费

> 整个发送和消费的过程清楚了，那么你就应该很容易想到上面这种情况如何实现了
> 
> 是不是只要我们将`3`条消息`A`、`B`、`C`发送到一个`MessageQueue`上，那么消费者消费就一定是按这个顺序消费的。
> 
> 如何实现将这三条消息`A`、`B`、`C`发送到一个`MessageQueue`上呢？

## 4.1 顺序消息生产 ##

public class Producer { 
    
        public static void main(String[] args) { 
            DefaultMQProducer producer = new DefaultMQProducer("fifo_test");
            producer.setNamesrvAddr("localhost:9876");
            try { 
                producer.start();
                // 将100条消息按照除16余数的规则，分别发送到对应索引的MessageQueue中
                for (int i = 0; i < 100; i++) { 
                    Message message = new Message();
                    message.setTopic("test_topic");
                    message.setTags("tagA");
                    message.setBody(new Integer(i).toString().getBytes(StandardCharsets.UTF_8));
                    SendResult sendResult =producer.send(message, new MessageQueueSelector() { 
                        @Override
                        public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) { 
                            int id= (int) arg;
                            int index = id % mqs.size();
                            // 根据余数来判断消息投放的MessageQueue
                            return mqs.get(index);
                        }
                    },i);
                    System.out.printf("%s%n", sendResult);
                }
            } catch (Exception e) { 
                e.printStackTrace();
            }
        }
    }

> 根据余数来判断消息投放的`MessageQueue`

## 4.2 顺序消息消费 ##

public class Consumer { 
        public static void main(String[] args) { 
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_test");
            consumer.setNamesrvAddr("localhost:9876");
            try { 
                consumer.subscribe("test_topic","tagA");
                // 可以看到每个queue有唯一的consume线程来消费, 对每个queue(分区)有序
                consumer.registerMessageListener(new MessageListenerOrderly() { 
                    @Override
                    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) { 
                        ListIterator<MessageExt> iterator = msgs.listIterator();
                        while (iterator.hasNext()){ 
                            MessageExt next = iterator.next();
                            try { 
                                System.out.println(Thread.currentThread().getName()+":"
                                +new String(next.getBody(),"utf-8")
                                        +"queueId=" + next.getQueueId() );
                            } catch (UnsupportedEncodingException e) { 
                                e.printStackTrace();
                            }
                        }
                        return ConsumeOrderlyStatus.SUCCESS;
                    }
                });
                consumer.start();
            } catch (MQClientException e) { 
                e.printStackTrace();
            }
        }
    }

> 每一个队列都都有一个唯一的线程来进行消费，并且在每一个`MessageQueue`中，先发的消息先消费

# 5 延时消息 #

其实就是`rocketmq`收到消息后，不会立刻推送给消费者，而是要等待一段时间（由用户设置）再推送给消费者消费。

用法很简单，只需要在创建消息的时候指定消息的延时级别就行

public class Producer { 
    
        public static void main(String[] args) { 
            DefaultMQProducer producer = new DefaultMQProducer("producer_test");
            producer.setNamesrvAddr("localhost:9876");
            try { 
                producer.start();
                Message message = new Message();
                message.setTopic("test_topic");
                message.setTags("tagA");
                message.setBody(("这是一条延时消息:"+ LocalDate.now().toString()).getBytes(StandardCharsets.UTF_8));
                // 设置延时等级3,这个消息将在10s之后发送消费者(现在只支持固定的几个时间,详看delayTimeLevel)
                message.setDelayTimeLevel(4);
                SendResult sendResult = producer.send(message);
                System.out.printf("%s%n", sendResult);
            } catch (MQClientException e) { 
                e.printStackTrace();
            } catch (InterruptedException e) { 
                e.printStackTrace();
            } catch (RemotingException e) { 
                e.printStackTrace();
            } catch (MQBrokerException e) { 
                e.printStackTrace();
            }
        }
    }

消费者的代码和普通消费者一样，这里就不再贴出来了了

[Link 1]: https://github.com/apache/rocketmq/blob/master/docs/cn/RocketMQ_Example.md