比较耗时而且不需要即时（同步）返回结果的操作作为消息放入消息队列 。同时，由于使用了消息队列MQ，只要保证消息格式不变，消息的发送方和接收方并不需要彼此联系，也不需要受对方的影响，即解耦合。 秒杀或团队抢购(高并发)活动中 使用广泛。 用户发起海量秒杀请求1到秒杀业务处理系统。 秒杀处理系统按照秒杀处理逻辑将满足秒杀条件的请求1发送到消息队列MQ。 下游的通知系统订阅1消息队列MQ的秒杀相关消息，再将1秒杀成功的消息发送到相关用户。 用户收到秒杀成功的通知。 下载 我们先从官网去选择Binary下载，Binary代表是二进制版本，可以直接使用。 解压 bin：启动脚本，包括shell脚本和CMD脚本 conf：实例配置文件 ，包括broker配置文件、logback配置文件等 lib：依赖jar包，包括Netty、commons-lang、FastJSON等 配置环境环境：ROCKETMQ_HOME 在电脑高级配置（配置jdk那边）配置环境变量，变量名为：ROCKETMQ_HOME，值为：MQ解压路径\MQ文件夹名。 修改配置文件 进入RocketMQ的conf文件，里面有一个broker.conf配置文件。在文件末尾添加两行。（输入的时候建议将中文去掉） enablePropertyFilter=true <!--支持标签过滤-->
启动nameserver 在MQ解压文件的bin目录下进入CMD的1命令行，执执行命令（或者双击 mqnamesrv.cmd ），启动nameserver， 成功后会弹出提示框，不要关闭。 # 执行命令
启动broker 在bin目录下cmd进入命令行模式，执行命令，启动broker。 成功后不要关闭 。 mqbroker.cmd -c ../conf/broker.conf
MQ可视化工具安装 可视化工具下载地址，密码6666 这个jar一定要JDK8的环境 ！！下载后修改application.properties文件，修改自己启动的端口和服务的ip+端口。集群按钮1那里出现了版本号说明启动成功！ 上传RocketMQ RocketMQ的下载包：密码6666 unzip /usr/soft/rocketmq-all-4.4.0-bin-release.zip # 这个目录要改为自己的RocketMQ
修改配置文件 进入RocketMQ的conf目录，修改broker.conf，新增三行(去掉我注释的中文)。 enablePropertyFilter=true # 支持标签顾过滤消息
修改启动参数 因为默认的RocketMQ启动的时候需要的1JVM内存很多，会报错，所以我们将他变小一点即可。默认是1G，修改为512m。 vi /usr/soft/rocketmq-4.4/bin/runbroker.sh
防火墙和安全组开放几个端口 开放9876、10911、10912、10909以及你指定的可视化工具启动的端口，这里步骤不再赘述，有疑问可以在评论区回复或者翻我的Linux的学习文章进行学习。 启动服务 nohup sh /usr/soft/rocketmq-4.4/bin/mqnamesrv &
查看日志 在RocketMQ的根目录的bin目录中，有一个nohup.out用于存放启动的日志信息。无报错即成功，日志信息和1Windows的差不多。 名字服务Name Server（邮局）：消息队列的协调者，Broker向它注册路由信息，同时Producer和Consumer向其获取路由信息，名称服务充当路由消息的提供者。生产者或消费者能够通过名字服务查找各主题相应的Broker IP列表。多个Namesrv实例组成集群，但相互独立，没有信息交换。 代理服务器Broker Server（邮递员）： Broker是RocketMQ的核心 ，消息中转角色，负责存储消息、转发消息。代理服务器在RocketMQ系统中负责接收从生产者发送来的消息并存储、同时为消费者的拉取请求作准备。代理服务器也存储消息相关的元数据，包括消费者组、消费进度偏移和主题和队列消息等。 生产者Producer（寄件人）：负责生产消息，一般由业务系统负责生产消息。一个消息生产者会把业务应用系统里产生的消息发送到broker服务器。RocketMQ提供多种发送方式，同步发送、异步发送、顺序发送、单向发送。同步和异步方式均需要Broker返回确认信息，单向发送不需要。 消费者Consumer（收件人）：负责消费消息，一般是后台系统负责异步消费。一个消息消费者会从Broker服务器拉取消息、并将其提供给应用程序。从用户应用的角度而言提供了两种消费形式：拉取式消费、推动式消费。 Topic（省份）：用来区分不同类型的消息，表示一类消息的集合，每个主题包含若干条消息，每条消息只能属于一个主题，是RocketMQ进行消息订阅的基本单位。 标签Tag：为消息设置的标志，用于同一主题下区分不同类型的消息。来自同一业务单元的消息，可以根据不同业务目的在同一主题下设置不同标签。标签能够有效地保持代码的清晰度和连贯性，并优化RocketMQ提供的查询系统。消费者可以根据Tag实现对不同子主题的不同消费逻辑，实现更好的扩展性。 消息队列MessageQueue(邮件)：为了提高性能和吞吐量，引入了Message Queue，一个Topic可以设置一个或多个Message Queue，这样消息就可以并行往各个Message Queue发送消息，消费者也可以并行的从多个 Message Queue读取消息 消息内容Message：息系统所传输信息的物理载体，生产和消费数据的最小单位，每条消息必须属于一个主题。RocketMQ中每个消息拥有唯一的Message ID，且可以携带具有业务标识的Key。系统提供了通过Message ID和Key查询消息的功能。 生产者组Producer Group：生产者组，简单来说就是多个发送同一类消息的生产者称之为一个生产者组。 消费者组Consumer Group：消费者组，消费同一类消息的多个 consumer 实例组成一个消费者组。 <dependency>
创建消息生产者, 指定生产者所属的组名。 指定Nameserver地址。 启动生产者。 创建消息对象，指定主题、标签和消息体。 发送消息。 关闭生产者。 public class Producer {public static void main(String[] args) throws Exception {// 创建一个生产者对象，并且指定一个生产者组 DefaultMQProducer producer = new DefaultMQProducer("xiaolin-producer");// 指定名字服务器地址 producer.setNamesrvAddr("127.0.0.1:9876");// 启动生产者 producer.start();// 创建一个消息，参数分别为：主题、标签、消息体 Message message = new Message("hello_demo1","tag1","你好,我是消息1".getBytes("utf-8"));// 发送消息 producer.send(message);// 关闭资源 producer.shutdown();
public class ConsumerDemo {public static void main(String[] args) throws MQClientException {// 创建一个拉取消息对象，并指定所属组名 DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer-demo");// 指定名字服务器地址 producer.setNamesrvAddr("127.0.0.1:9876");// 指定消费者订阅的主题和标签，第二个参数用于过滤条件用于指定接收什么tag，什么都接收 consumer.subscribe("hello_demo1", "*");// 注册一个消息监听器 consumer.registerMessageListener(new MessageListenerConcurrently() {public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,// 可能有多个消息 for (MessageExt msg : msgs) {// 打印消息 System.out.println(new String(msg.getBody()));// 返回发送成功 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;// 启动消费者 consumer.start();
public class SyncProducer {public static void main(String[] args) throws Exception {// 实例化消息生产者Producer DefaultMQProducer producer = new DefaultMQProducer("xiaolin-producer");// 设置NameServer的地址 producer.setNamesrvAddr("127.0.0.1:9876");// 启动Producer实例 producer.start();for (int i = 0; i < 100; i++) {// 创建消息，并指定Topic，Tag和消息体 Message msg = new Message("04-producer-type" /* Topic */ ,/* Tag */ ,/* Message body */ );//发送同步消息到一个Broker SendResult sendResult = producer.send(msg);// 通过sendResult返回消息是否成功送达 System.out.println(JSON.toJSONString(sendResult));// 如果不再发送消息，关闭Producer实例。 producer.shutdown();
public class ASyncProducer {public static void main(String[] args) throws Exception {// 实例化消息生产者Producer DefaultMQProducer producer = new DefaultMQProducer("xiaolin-producer");// 设置NameServer的地址 producer.setNamesrvAddr("127.0.0.1:9876");// 启动Producer实例 producer.start();for (int i = 0; i < 100; i++) {// 创建消息，并指定Topic，Tag和消息体 Message msg = new Message("04-producer-type" /* Topic */ ,/* Tag */ ,/* Message body */ //发送同步消息到一个Broker producer.send(msg, new SendCallback() {public void onSuccess(SendResult sendResult) {public void onException(Throwable e) {// 让线程不要终止，否则会报错 Thread.sleep(30000000);// 如果不再发送消息，关闭Producer实例。 producer.shutdown();
public class OneWayProducer {public static void main(String[] args) throws Exception {// 实例化消息生产者Producer DefaultMQProducer producer = new DefaultMQProducer("xiaolin-producer");// 设置NameServer的地址 producer.setNamesrvAddr("127.0.0.1:9876");// 启动Producer实例 producer.start();for (int i = 0; i < 100; i++) {// 创建消息，并指定Topic，Tag和消息体 Message msg = new Message("04-producer-type" /* Topic */ ,/* Tag */ ,/* Message body */ //发送单向消息到一个Broker producer.sendOneway(msg);// 如果不再发送消息，关闭Producer实例。 producer.shutdown();
[/code][code] public class Consumer {public static void main(String[] args) throws MQClientException {//1.创建消费者Consumer，制定消费者组名 DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");//2.指定Nameserver地址 consumer.setNamesrvAddr("192.168.25.135:9876;192.168.25.138:9876");//3.订阅主题Topic和Tag consumer.subscribe("OrderTopic", "*");//4.注册消息监听器 consumer.registerMessageListener(new MessageListenerOrderly() {public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {for (MessageExt msg : msgs) {new String(msg.getBody()));return ConsumeOrderlyStatus.SUCCESS;//5.启动消费者 consumer.start();
半事务消息：暂不能投递的消息，发送方已经成功地将消息发送到了RocketMQ服务端，但是服务端未收到生产者对该消息的二次确认，此时该消息被标记成“暂不能投递”状态，处于该种状态下的消息即半事务消息。 消息回查：由于网络闪断、生产者应用重启等原因，导致某条事务消息的二次确认丢失，RocketMQ服务端通过扫描发现某条消息长期处于“半事务消息”时，需要主动向消息生产者询问该消息的最终状态（Commit 或是 Rollback），该询问过程即消息回查。 发送方将半事务消息发送到RocketMQ服务端。 RocketMQ服务端将消息持久化后，向发送方返回确认消息已经发送成功，此时消息为半事务消息。 发送方开始执行本地事务逻辑。 发送方根据本地事务执行结果向服务端提交二次确认（Commit或者是Rollback），服务端收到Commit状态则将事务消息标记为可投递，订阅方最终将收到该消息。服务端如果收到的是Rollback状态则删除半事务消息，订阅方将不会接收该消息。 在断网或者是应用重启的特殊情况下，上述步骤4提交的二次确认最终未到达服务端，经过固定时间后服务端将对该消息发起消息回查。 发送方收到消息回查后，需要检查对应消息的本地事务执行的最终结果。 发送方根据检查得到的本地事务的最终状态再次提交二次确认，服务端仍按照步骤4对半事务消息进行操作。 private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
public class ScheduledMessageProducer {public static void main(String[] args) throws Exception {// 实例化一个生产者来产生延时消息 DefaultMQProducer producer = new DefaultMQProducer("wolfcode-producer");// 启动生产者 producer.start();new Message("06-delay", ("delay message").getBytes());// 设置延时等级3,这个消息将在10s之后发送(现在只支持固定的几个时间,详看delayTimeLevel) message.setDelayTimeLevel(3);// 发送消息 producer.send(message);// 关闭生产者 producer.shutdown();
public class ScheduledMessageConsumer {public static void main(String[] args) throws Exception {new DefaultMQPushConsumer("wolfcode_consumer");// 订阅Topics consumer.subscribe("06-delay", "*");// 注册消息监听者 consumer.registerMessageListener(new MessageListenerConcurrently() {public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context) {for (MessageExt message : messages) {return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;// 启动消费者 consumer.start();
通过标签过滤。 通过SQL语句的方式过滤。 public class Producer {public static void main(String[] args) throws Exception {//1.创建消息生产者producer，并制定生产者组名 DefaultMQProducer producer = new DefaultMQProducer("group1");//2.指定Nameserver地址 producer.setNamesrvAddr("127.0.0.1:9876");//3.启动producer producer.start();for (int i = 0; i < 3; i++) {//4.创建消息对象，指定主题Topic、Tag和消息体 /** new Message("FilterTagTopic", "Tag2", ("消息的过滤" + i).getBytes());//5.发送消息 SendResult result = producer.send(msg);//发送状态 SendStatus status = result.getSendStatus();//线程睡1秒 TimeUnit.SECONDS.sleep(1);//6.关闭生产者producer producer.shutdown();
public class Consumer {public static void main(String[] args) throws Exception {//1.创建消费者Consumer，制定消费者组名 DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");//2.指定Nameserver地址 consumer.setNamesrvAddr("127.0.0.1:9876");//3.订阅主题Topic和Tag consumer.subscribe("FilterTagTopic", "Tag1 || Tag2 ");//4.设置回调函数，处理消息 consumer.registerMessageListener(new MessageListenerConcurrently() {//接受消息内容 @Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {for (MessageExt msg : msgs) {new String(msg.getBody()));return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//5.启动消费者consumer consumer.start();
数值比较，比如： >，>=，<，<=，BETWEEN，= 字符比较，比如： =，<>，IN IS NULL ?或者? IS NOT NULL 逻辑符号? AND，OR，NOT 数值，比如： 123，3.1415 字符，比如： 'abc'，必须用单引号包裹起来 NULL ，特殊的常量 布尔值， TRUE ?或? FALSE DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");new Message("TopicTest",// 设置一些属性 msg.putUserProperty("a", String.valueOf(i));
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4");// 只有订阅的消息有这个属性a, a >=0 and a <= 3 consumer.subscribe("TopicTest", MessageSelector.bySql("a between 0 and 3");new MessageListenerConcurrently() {public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
www.javathinker.net
网站系统异常
系统异常信息
Request URL:
http://www.javathinker.net/WEB-INF/lybbs/jsp/topic.jsp?postID=3926本站管理人员 。
Java技术网站 → 『 Java开发技术 』 → 『 Spring框架和各种开源软件技术专题 』 → 10分钟认识RocketMQ！想进阿里连这个都不会？
消息
查看
搜索
好友
邮件
复制
引用
