./bin/mqnamesrv

./bin/mqbroker -n {name.server.url} 
./bin/mqbroker -c {config}  
#-n 指定nameserver的ip和端口
#-c 以指定配置文件为参数启动
#端口默认为 9876

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

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
?
public class SyncProducer {
    public static void main(String[] args) throws Exception {
        //Instantiate with a producer group name.
        DefaultMQProducer producer = new DefaultMQProducer("test-group");
        // Specify name server addresses.
        producer.setNamesrvAddr("172.16.55.185:9876");
        //Launch the instance.
        producer.start();
        for (int i = 0; i < 100; i++) {
            //Create a message instance, specifying topic, tag and message body.
            Message msg = new Message("TopicTest11" /* Topic */,
                    "TagA" /* Tag */,
                    ("Hello RocketMQ " +
                            i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
            );
            //Call send message to deliver message to one of brokers.
            SendResult sendResult = producer.send(msg);
            System.out.printf("%s%n", sendResult);
        }
        //Shut down once the producer instance is not longer in use.
        producer.shutdown();
    }
}

DefaultMQProducer producer = new DefaultMQProducer("HAOKE_IM");
producer.setNamesrvAddr("172.16.55.185:9876");
producer.start();
/**
 * key：broker名称
 * newTopic：topic名称
 * queueNum：队列数（分区）默认4个
 */
producer.createTopic("broker_name", "topic", 8);

String msgStr = "用户A发送消息给用户B";
Message msg = new Message("haoke_im_topic","SEND_MSG",
msgStr.getBytes(RemotingHelper.DEFAULT_CHARSET));
// 发送消息
SendResult sendResult = producer.send(msg);
System.out.println("消息状态：" + sendResult.getSendStatus());
System.out.println("消息id：" + sendResult.getMsgId());
System.out.println("消息queue：" + sendResult.getMessageQueue());
System.out.println("消息offset：" + sendResult.getQueueOffset());

String msgStr = "用户A发送消息给用户B";
Message msg = new Message("broker_name","SEND_MSG",msgStr.getBytes(RemotingHelper.DEFAULT_CHARSET));
// 异步发送消息
producer.send(msg, new SendCallback() {
    @Override
    public void onSuccess(SendResult sendResult) {
        System.out.println("消息状态：" + sendResult.getSendStatus());
        System.out.println("消息id：" + sendResult.getMsgId());
        System.out.println("消息queue：" + sendResult.getMessageQueue());
        System.out.println("消息offset：" + sendResult.getQueueOffset());
    }
    @Override
    public void onException(Throwable e) {
        System.out.println("发送失败！" + e);
    }
});

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_group");
consumer.setNamesrvAddr("172.16.55.185:9876");
// 订阅topic，* 表示接收此Topic下的所有消息，相当于一个匹配符 
//如果为 "tag1"就表示只订阅tag为"tag1"的消息 还可以组合订阅 "tag1 || tag2 ||  tag3"
consumer.subscribe("topic_name", "*");
consumer.registerMessageListener(new MessageListenerConcurrently() {
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
    ConsumeConcurrentlyContext context) {
        for (MessageExt msg : msgs) {
            try {
                System.out.println(new String(msg.getBody(), "UTF-8"));
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
            }
        }
        System.out.println("收到消息->" + msgs);
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
});
consumer.start();

String msgStr = "大家好我叫小美，今年18岁";
Message msg = new Message("topic_name","tag1",
msgStr.getBytes(RemotingHelper.DEFAULT_CHARSET));
msg.putUserProperty("age", "18");
msg.putUserProperty("sex", "女");
// 发送消息
SendResult sendResult = producer.send(msg);

// 订阅topic，接收此Topic下的所有消息
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_group");
consumer.setNamesrvAddr("172.16.55.185:9876");
// 订阅topic，* 表示接收此Topic下的所有消息，相当于一个匹配符 
consumer.subscribe("topic_name", MessageSelector.bySql("age>=18 AND sex=‘女‘"));
consumer.registerMessageListener(new MessageListenerConcurrently() {
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
    ConsumeConcurrentlyContext context) {
        for (MessageExt msg : msgs) {
            try {
                System.out.println(new String(msg.getBody(), "UTF-8"));
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
            }
        }
        System.out.println("收到消息->" + msgs);
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
});
consumer.start();

DefaultMQProducer producer = new DefaultMQProducer("producer_group");
producer.setNamesrvAddr("172.16.55.185:9876");
producer.start();
for (int i = 0; i < 100; i++) {
    String msgStr = "order --> " + i;
    int orderId = i % 10; // 模拟生成订单id
    Message message = new Message("topic","tag",
                                msgStr.getBytes(RemotingHelper.DEFAULT_CHARSET));
    //orderId实际传入到lambda表达式（MessageQueueSelector）第三个参数
    SendResult sendResult = producer.send(message, (mqs, msg, arg) -> {
        //orderId范围0-9之间
        Integer id = (Integer) arg;
        //topic默认queue个数为4 所以index范围0-4之间
        int index = id % mqs.size();
        //凡是orderId对4取余为0的选用第1个queue
        //凡是orderId对4取余为1的选用第2个queue
        //凡是orderId对4取余为2的选用第3个queue
        //凡是orderId对4取余为3的选用第4个queue
        //这样一来就保证了满足某一条件的消息能够总是发送到特定的queue中
        return mqs.get(index);
    }, orderId);
    System.out.println(sendResult);
}
producer.shutdown();

DefaultMQPushConsumer consumer = new
DefaultMQPushConsumer("consumer_group");
consumer.setNamesrvAddr("172.16.55.185:9876");
consumer.subscribe("topic", "tag");
//使用MessageListenerOrderly这种顺序监听器，可以满足按顺序消费同一队列中的消息，不同的线程处理不同的队列
//比如thread-pool-1专门消费queue-0中的消息
//比如thread-pool-2专门消费queue-1中的消息
//比如thread-pool-3专门消费queue-2中的消息
//比如thread-pool-4专门消费queue-3中的消息
consumer.registerMessageListener(new MessageListenerOrderly() {
    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
        ConsumeOrderlyContext context) {
        System.out.println(Thread.currentThread().getName() + " Receive New
        Messages: " + msgs);
        return ConsumeOrderlyStatus.SUCCESS;
    }
});
consumer.start();

TransactionMQProducer producer = new
TransactionMQProducer("producer_group");
producer.setNamesrvAddr("172.16.55.185:9876");
// 设置自定义事务监听器
producer.setTransactionListener(new TransactionListenerImpl());
producer.start();
// 发送消息
Message message = new Message("pay_topic", "用户A给用户B转账500元".getBytes("UTF-8"));
producer.sendMessageInTransaction(message, null);
Thread.sleep(999999);
producer.shutdown();

public class TransactionListenerImpl implements TransactionListener {
    private static Map<String, LocalTransactionState> STATE_MAP = new HashMap<>();
    /**
    * 执行具体的业务逻辑
    *
    * @param msg 发送的消息对象
    * @param arg
    * @return
    */
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        try {
            System.out.println("用户A账户减500元.");
            Thread.sleep(500); //模拟调用服务
            // System.out.println(1/0);
            System.out.println("用户B账户加500元.");
            Thread.sleep(800);
            STATE_MAP.put(msg.getTransactionId(),
            LocalTransactionState.COMMIT_MESSAGE);
            // 二次提交确认
            return LocalTransactionState.COMMIT_MESSAGE;
        } catch (Exception e) {
            e.printStackTrace();
        }
        STATE_MAP.put(msg.getTransactionId(),
        LocalTransactionState.ROLLBACK_MESSAGE);
        // 回滚
        return LocalTransactionState.ROLLBACK_MESSAGE;
    }
    /**
    * 消息回查
    *
    * @param msg
    * @return
    */
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        return STATE_MAP.get(msg.getTransactionId());
    }
}

// 集群模式
consumer.setMessageModel(MessageModel.CLUSTERING);
// 广播模式
consumer.setMessageModel(MessageModel.BROADCASTING);

#-- 异步
flushDiskType = ASYNC_FLUSH 
#-- 同步
flushDiskType = SYNC_FLUSH