mirror of
https://github.com/apache/rocketmq.git
synced 2026-09-01 15:47:01 +08:00
136 lines
5.2 KiB
Markdown
136 lines
5.2 KiB
Markdown
# Basic Sample
|
|
------
|
|
基本示例中提供了以下两个功能
|
|
* RocketMQ可用于以三种方式发送消息:可靠的同步、可靠的异步和单向传输。前两种消息类型是可靠的,因为无论它们是否成功发送都有响应。
|
|
* RocketMQ可以用来消费消息。
|
|
### 1 添加依赖
|
|
maven:
|
|
``` java
|
|
<dependency>
|
|
<groupId>org.apache.rocketmq</groupId>
|
|
<artifactId>rocketmq-client</artifactId>
|
|
<version>4.3.0</version>
|
|
</dependency>
|
|
```
|
|
gradle:
|
|
``` java
|
|
compile 'org.apache.rocketmq:rocketmq-client:4.3.0'
|
|
```
|
|
### 2 发送消息
|
|
##### 2.1 使用Producer发送同步消息
|
|
可靠的同步传输被广泛应用于各种场景,如重要的通知消息、短消息通知等。
|
|
``` java
|
|
public class SyncProducer {
|
|
public static void main(String[] args) throws Exception {
|
|
// Instantiate with a producer group name
|
|
DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
|
|
// Specify name server addresses
|
|
producer.setNamesrvAddr("localhost:9876");
|
|
// Launch the producer instance
|
|
producer.start();
|
|
for (int i = 0; i < 100; i++) {
|
|
// Create a message instance with specifying topic, tag and message body
|
|
Message msg = new Message("TopicTest" /* Topic */,
|
|
"TagA" /* Tag */,
|
|
("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
|
|
);
|
|
// Send message to one of brokers
|
|
SendResult sendResult = producer.send(msg);
|
|
// Check whether the message has been delivered by the callback of sendResult
|
|
System.out.printf("%s%n", sendResult);
|
|
}
|
|
// Shut down once the producer instance is not longer in use
|
|
producer.shutdown();
|
|
}
|
|
}
|
|
```
|
|
##### 2.2 发送异步消息
|
|
异步传输通常用于响应时间敏感的业务场景。这意味着发送方无法等待代理的响应太长时间。
|
|
``` java
|
|
public class AsyncProducer {
|
|
public static void main(String[] args) throws Exception {
|
|
// Instantiate with a producer group name
|
|
DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
|
|
// Specify name server addresses
|
|
producer.setNamesrvAddr("localhost:9876");
|
|
// Launch the producer instance
|
|
producer.start();
|
|
producer.setRetryTimesWhenSendAsyncFailed(0);
|
|
for (int i = 0; i < 100; i++) {
|
|
final int index = i;
|
|
// Create a message instance with specifying topic, tag and message body
|
|
Message msg = new Message("TopicTest",
|
|
"TagA",
|
|
"OrderID188",
|
|
"Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
|
|
// SendCallback: receive the callback of the asynchronous return result.
|
|
producer.send(msg, new SendCallback() {
|
|
@Override
|
|
public void onSuccess(SendResult sendResult) {
|
|
System.out.printf("%-10d OK %s %n", index,
|
|
sendResult.getMsgId());
|
|
}
|
|
@Override
|
|
public void onException(Throwable e) {
|
|
System.out.printf("%-10d Exception %s %n", index, e);
|
|
e.printStackTrace();
|
|
}
|
|
});
|
|
}
|
|
// Shut down once the producer instance is not longer in use
|
|
producer.shutdown();
|
|
}
|
|
}
|
|
```
|
|
##### 2.3 以单向模式发送消息
|
|
单向传输用于需要中等可靠性的情况,如日志收集。
|
|
``` java
|
|
public class OnewayProducer {
|
|
public static void main(String[] args) throws Exception{
|
|
// Instantiate with a producer group name
|
|
DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
|
|
// Specify name server addresses
|
|
producer.setNamesrvAddr("localhost:9876");
|
|
// Launch the producer instance
|
|
producer.start();
|
|
for (int i = 0; i < 100; i++) {
|
|
// Create a message instance with specifying topic, tag and message body
|
|
Message msg = new Message("TopicTest" /* Topic */,
|
|
"TagA" /* Tag */,
|
|
("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
|
|
);
|
|
// Send in one-way mode, no return result
|
|
producer.sendOneway(msg);
|
|
}
|
|
// Shut down once the producer instance is not longer in use
|
|
producer.shutdown();
|
|
}
|
|
}
|
|
```
|
|
### 3 消费消息
|
|
``` java
|
|
public class Consumer {
|
|
public static void main(String[] args) throws InterruptedException, MQClientException {
|
|
// Instantiate with specified consumer group name
|
|
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");
|
|
|
|
// Specify name server addresses
|
|
consumer.setNamesrvAddr("localhost:9876");
|
|
|
|
// Subscribe one or more topics and tags for finding those messages need to be consumed
|
|
consumer.subscribe("TopicTest", "*");
|
|
// Register callback to execute on arrival of messages fetched from brokers
|
|
consumer.registerMessageListener(new MessageListenerConcurrently() {
|
|
@Override
|
|
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
|
|
System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
|
|
// Mark the message that have been consumed successfully
|
|
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
|
|
}
|
|
});
|
|
// Launch the consumer instance
|
|
consumer.start();
|
|
System.out.printf("Consumer Started.%n");
|
|
}
|
|
}
|
|
``` |