RocketMQ入门案例

MQ作用（优点）:

• 1、应用解耦（异步消息发送）基础
• 2、快速应用变更维护
• 3、流量削峰
  缺点：
• 1、系统的可用性降低（通过集群解决MQ宕机的问题）
  • （服务A和服务B不是直接通信，A和B直接需要通过MQ）【MQ类似于媒婆】
• 2、系统复杂度提高
• 3、异步消息机制
  • 消息的顺序性（将所有消息丢给MQ，顺序不能保证）
  • 消息丢失（MQ可能宕机，导致数据丢失）
  • 消息一致性
  • 消息重复使用的问题

常见MQ产品：
ActiveMQ：java语言实现，万级数据吞吐量，处理速度ms级，主从架构，成熟度高。
RabbitMQ：erlang语言实现，万级数据吞吐量，处理速度us级，主从架构。
RocketMQ：Java语言实现，十万及数据吞吐量，处理速度ms级，分布式架构，功能强大，扩展性强。
kafka：scala语言实现，十万计吞吐量，处理速度ms级，分布式架构，功能较少，应用于大数据较多

RocketMQ的安装：

• 1、安装JDK（1.8）
• 2、上传压缩包【npm -y install lrzsz】【rz】
• 3、解压【unzip rocketmq-all-4.5.2-bin-release.zip】unzip 压缩包文件名
• 4、修改目录名称【mv rocketmq-all-4.5.2-bin-release rocketmq】
  启动服务器：
• 1、启动命名服务器（bin目录下）
  sh mqnamesrv
• 2、启动消息服务器（bin目录下）
  sh mqbroker -n localhost:9876
  修改runbroker.sh文件中的内存配置
  调整与当前虚拟机内存匹配，推荐256m、128m

  org.apache.rocketmq
  rocketmq-client
  4.5.2

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

public class Producer {
	
	public static void main(String[] args) {
		// 创建发送消息的对象（不传参默认有一个分组，传参则需要一个组名）
		DefaultMQProducer producer = new DefaultMQProducer(Constant.DEFAULT_GROUP);
		try {
			// 设置发送的命名服务器地址
			producer.setNamesrvAddr(Constant.NAME_SERVER_ADDR);
			// 启动发送服务
			producer.start();
			// 设置超时时间
			producer.setSendMsgTimeout(10000);
			// 创建消息对象，指定topic主题，指定body消息内容
			byte[] body = "HELLO ROCKETMQ!".getBytes(Constant.CHAR_SET);
			Message message = new Message(Constant.TOPIC_STR, body);
			// 发送消息方法
			SendResult result = producer.send(message, 0L);
			// 打印返回结果
			System.out.println(result);
		} catch (Exception e) {
			System.out.println("======出现异常信息======");
			System.out.println(e.getMessage());
			e.printStackTrace();
		} finally {
			// 关闭连接
			producer.shutdown();
		}
	}
}

import java.util.List;
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.common.message.MessageExt;

public class Consumer {
	public static void main(String[] args) {
		// 创建接收的消息对象（不传参默认有一个分组，传参则需要一个组名）
		DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(Constant.DEFAULT_GROUP);
		try {
			// 设定接收的命名服务器地址（与消息发送者的地址保持一致）
			consumer.setNamesrvAddr(Constant.NAME_SERVER_ADDR);
			// 设置接收消息对应的topic标题，对应的sub标签为任意（producer中未定义，这里匹配任意类型）
			consumer.subscribe(Constant.TOPIC_STR, "*");
			// 开启监听，监听消息数据
			consumer.registerMessageListener(new MessageListenerConcurrently() {
				@Override
				public ConsumeConcurrentlyStatus consumeMessage(List msgs,
						ConsumeConcurrentlyContext context) {
					msgs.forEach(item -> {
						System.out.println("接收到消息：" + item);
						System.out.println("接收到消息内容：" + new String(item.getBody()));
						System.out.println(item.getBornHostNameString());
						System.out.println(item.getBornHostString());
					});
					return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
				}
			});
			// 开启消息接收的服务
			consumer.start();
			System.out.println("接收消息的服务已开始运行！");
		} catch (Exception e) {
			System.out.println("======出现异常信息======");
			System.out.println(e.getMessage());
			e.printStackTrace();
		} finally {
			consumer.shutdown();
		}
	}
}

public class Constant {
	
	public static final String NAME_SERVER_ADDR = "127.0.0.1:9876";
	
	public static final String TOPIC_STR = "TOPIC_1";
	
	public static final String CHAR_SET = "UTF-8";
	
	public static final String DEFAULT_GROUP = "GROUP_1";
}