RocketMQ——IDEA开发RocketMQ程序简单入门详解编程语言

建立maven项目，导入下面依赖

<!-- https://mvnrepository.com/artifact/org.apache.rocketmq/rocketmq-client --> 
        <dependency> 
            <groupId>org.apache.rocketmq</groupId> 
            <artifactId>rocketmq-client</artifactId> 
            <version>4.6.1</version> 
        </dependency> 

保证已在服务器启动nameserver和broker，强烈推荐读者使用上文提到的网页版UI观察消息队列状态，下面展示三种发送生产消费的方法：

import org.apache.rocketmq.client.producer.DefaultMQProducer; 
import org.apache.rocketmq.client.producer.SendResult; 
import org.apache.rocketmq.common.message.Message; 
 
/** 
 * 消息发送者Producer 
 * 
 * @author Song X. 
 * @date 2020/03/02 
 */ 
public class Producer_Sync {
    
    public static void main(String[] args) throws Exception {
    
        //创建生产者组 
        DefaultMQProducer producer = new DefaultMQProducer("ProducerGroup01"); 
        //设置nameserver地址，如1.1.1.1:9876 
        producer.setNamesrvAddr("enter the addr of nameserver"); 
        //启动producer 
        producer.start(); 
 
        //topic消息将要发往的地址 
        //body是真正的消息体 
        Message msg = new Message("myTopic001", "RocketMQ 第一条消息".getBytes()); 
 
        //同步阻塞式发消息，也就是发送完消息，必须等broker返回一个发送成功的消息，才会做其他事 
        //这种方式消息发送的安全性可以得到保证，但是同步阻塞必定慢 
        SendResult sendResult = producer.send(msg); 
        System.out.println(sendResult); 
         
        //关闭消费者 
        producer.shutdown(); 
    } 
} 
 

import org.apache.rocketmq.client.producer.DefaultMQProducer; 
import org.apache.rocketmq.client.producer.SendCallback; 
import org.apache.rocketmq.client.producer.SendResult; 
import org.apache.rocketmq.common.message.Message; 
import org.apache.rocketmq.remoting.common.RemotingHelper; 
 
/** 
 * 消息发送者Producer 
 * 
 * @author Song X. 
 * @date 2020/03/02 
 */ 
public class Producer_Asyc {
    
    public static void main(String[] args) throws Exception {
    
        //创建生产者组 
        DefaultMQProducer producer = new DefaultMQProducer("ProducerGroup01"); 
        //设置nameserver地址 
        producer.setNamesrvAddr("enter the addr of nameserver"); 
        //启动producer 
        producer.start(); 
        //设置异步发送失败时的重设时间 
        producer.setRetryTimesWhenSendAsyncFailed(1); 
 
        //异步非阻塞式发消息，也就是发送完消息，不等broker返回一个发送成功的消息，就做其他事了 
        //那么如何获得broker的确认信息，采用监听的形式 
        //这种方式消息发送的安全性无法得到保证，但是速度快 
        for(int i = 0; i < 100; i++) {
    
            final int index = i; 
            StringBuffer body = new StringBuffer("Rocket MQ 第"); 
            body.append(i+"条消息"); 
 
            Message msg = new Message("myTopic001", body.toString().getBytes( 
                    RemotingHelper.DEFAULT_CHARSET 
            )); 
 
            producer.send(msg, new SendCallback() {
    
                @Override 
                public void onSuccess(SendResult sendResult) {
    
                    System.out.println("第 " + index + " 消息发送成功"); 
                } 
 
                @Override 
                public void onException(Throwable throwable) {
    
                    System.out.println("第 " + index + " 消息发送失败"); 
                    throwable.printStackTrace(); 
                } 
            }); 
        } 
 
        //关闭消费者，发送异步消息时，切勿使用这句话，因为无法确定上面的回调方法什么时候执行 
        //producer.shutdown(); 
    } 
} 

import org.apache.rocketmq.client.producer.DefaultMQProducer; 
import org.apache.rocketmq.client.producer.SendResult; 
import org.apache.rocketmq.common.message.Message; 
 
/** 
 * 消息发送者Producer 
 * 
 * @author Song X. 
 * @date 2020/03/02 
 */ 
public class Producer_OneWay {
    
    public static void main(String[] args) throws Exception {
    
        //创建生产者组 
        DefaultMQProducer producer = new DefaultMQProducer("ProducerGroup01"); 
        //设置nameserver地址 
        producer.setNamesrvAddr("192.168.255.131:9876"); 
        //启动producer 
        producer.start(); 
 
        //topic消息将要发往的地址 
        //body是真正的消息体 
        Message msg = new Message("myTopic001", "RocketMQ 第一条消息".getBytes()); 
 
        //单向发送消息，不关心有没有发送成功。速度最快 
        producer.sendOneway(msg); 
 
        //关闭消费者 
        producer.shutdown(); 
    } 
} 
 

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; 
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext; 
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; 
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; 
import org.apache.rocketmq.client.exception.MQClientException; 
import org.apache.rocketmq.common.message.MessageExt; 
 
import java.util.List; 
 
/** 
 * 消息消费者Consumer 
 * 
 * @author Song X. 
 * @date 2020/03/02 
 */ 
public class Consumer {
    
    public static void main(String[] args) throws MQClientException {
    
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroup01"); 
 
        //设置nameserver地址 
        consumer.setNamesrvAddr("enter the addr of nameserver"); 
 
        //每个consumer只能关注一个topic 
        //topic 关注的topic名 
        //subExpression 根据Tag过滤 表示我需要过滤掉当前topic里的什么信息, *代表不过滤 
        String topic = "myTopic001"; 
        String subExpression = "*"; 
        consumer.subscribe(topic, subExpression); 
 
        //使用监听的方式接受信息，防止消费者本身一直在这阻塞等着消息 
        consumer.registerMessageListener(new MessageListenerConcurrently() {
    
            //list就是收到的所有消息 
            @Override 
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
    
                for(MessageExt msg : list){
    
                    System.out.println(new String(msg.getBody())); 
                } 
 
                //给broker返回消费成功的信息,broker就会将这条消息状态改为success，这样这条消息之后就不会被其他consumer消费 
                //相当于给broker返回了一个ack信息 
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; 
            } 
        }); 
 
        //启动消费者 
        consumer.start(); 
        System.out.println("consumer start...."); 
    } 
}