|
# Basic Sample
|
||
|
|
------
|
||
|
|
Two functions below are provided in the basic sample:
|
||
|
|
* The RocketMQ can be utilized to send messages in three ways: reliable synchronous, reliable asynchronous, and one-way transmission. The first two message types are reliable because there is a response whether they were sent successfully.
|
||
|
|
* The RocketMQ can be utilized to consume messages.
|
||
|
|
### 1 Add Dependency
|
||
|
|
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 Send Messages
|
||
|
|
##### 2.1 Use Producer to Send Synchronous Messages
|
||
|
|
Reliable synchronous transmission is used in extensive scenes, such as important notification messages, SMS notification.
|
||
|
|
``` 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 Send Asynchronous Messages
|
||
|
|
Asynchronous transmission is generally used in response time sensitive business scenarios. It means that it is unable for the sender to wait the response of the Broker too long.
|
||
|
|
``` 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 Send Messages in One-way Mode
|
||
|
|
One-way transmission is used for cases requiring moderate reliability, such as log collection.
|
||
|
|
``` 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 Consume Messages
|
||
|
|
``` 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");
|
||
|
|
}
|
||
|
|
}
|
||
|
|
```
|