文档中心

		

			<!-- in your <dependencies> block -->
		

		

			 <dependency>
		

		

			 <groupId>org.apache.rocketmq</groupId>
		

		

			 <artifactId>rocketmq-client</artifactId>
		

		

			 <version>4.9.7</version>
		

		

			 </dependency>
		

		

			 
		

		

			 <dependency>
		

		

			 <groupId>org.apache.rocketmq</groupId>
		

		

			 <artifactId>rocketmq-acl</artifactId>
		

		

			 <version>4.9.7</version>
		

		

			 </dependency>
		

		

			// 实例化消息生产者Producer
		

		

			 DefaultMQProducer producer = new DefaultMQProducer(
		

		

			 groupName,
		

		

			 new AclClientRPCHook(new SessionCredentials(accessKey, secretKey)) // ACL权限
		

		

			 );
		

		

			 // 设置NameServer的地址,地址就是形如xxx.tencenttdmq.com:8080 这样的接入地址。
		

		

			 producer.setNamesrvAddr(nameserver);
		

		

			 // 启动Producer实例
		

		

			 producer.start();
		

		

			 
		

		

			 for (int i = 0; i < 10; i++) {
		

		

			 // 创建消息实例，设置topic和消息内容.
		

		

			 Message msg = new Message(topic_name, ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
		

		

			 // 发送消息
		

		

			 SendResult sendResult = producer.send(msg);
		

		

			 System.out.printf("%s%n", sendResult);
		

		

			 }
		

		

			// 实例化消费者
		

		

			DefaultMQPushConsumer pushConsumer = new DefaultMQPushConsumer(
		

		

			 groupName, 
		

		

			 new AclClientRPCHook(new SessionCredentials(accessKey, secretKey))); //ACL权限
		

		

			 // 设置NameServer的地址
		

		

			pushConsumer.setNamesrvAddr(nameserver);
		

		

			 // 订阅topic
		

		

			pushConsumer.subscribe(topic_name, "*");
		

		

			 // 注册回调实现类来处理从broker拉取回来的消息
		

		

			pushConsumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
		

		

			 // 消息处理逻辑
		

		

			 System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
		

		

			 // 标记该消息已经被成功消费, 根据消费情况，返回处理状态
		

		

			 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
		

		

			 });
		

		

			 // 启动消费者实例
		

		

			pushConsumer.start();