springbootdemo

To use RocketMQ with Spring Boot, you must first install and run the RocketMQ server components (NameServer and Broker). This guide uses version 4.3.0.

1. Build from Source

Download the source release, unzip it, and build using Maven:

# wget http://mirrors.hust.edu.cn/apache/rocketmq/4.3.0/rocketmq-all-4.3.0-source-release.zip
# unzip rocketmq-all-4.3.0-source-release.zip
# mvn -Prelease-all -DskipTests clean install -U
# cd distribution/target/apache-rocketmq

2. Start NameServer

The NameServer manages broker metadata. Ensure port 9876 is available.

# nohup sh bin/mqnamesrv &
# tail -f ~/logs/rocketmqlogs/namesrv.log

Look for The Name Server boot success... in the logs.

3. Start Broker

Start the Broker and point it to your NameServer address.

# nohup sh bin/mqbroker -n localhost:9876 &
# tail -f ~/logs/rocketmqlogs/broker.log

Look for The broker[...] boot success... in the logs.

The SpringBootDemo project is a demonstration of a technology stack integrating SpringBoot, MyBatis, Redis, and MySQL.

To run the project:

1. Clone the repository to your local machine.
2. Set up a MySQL database using the provided schema.
3. Configure the application properties to point to your local Redis and MySQL instances.
4. Run the application.

The code and configuration files include comments to guide you through the setup.

-- ----------------------------
-- Table structure for `user_info`
-- ----------------------------
DROP TABLE IF EXISTS `user_info`;
CREATE TABLE `user_info` (
  `id` int(8) NOT NULL AUTO_INCREMENT,
  `name` varchar(20) NOT NULL,
  `age` int(2) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8;

-- ----------------------------
-- Records of user_info
-- ----------------------------
INSERT INTO `user_info` VALUES ('1', 'xiaour', '18');

To send messages, create a @Component that initializes a DefaultMQProducer. Use @Value to inject configuration from your properties file.

Key steps:

1. Initialize DefaultMQProducer with a producerGroup.
2. Set the namesrvAddr.
3. Call producer.start() in a @PostConstruct method.
4. Use producer.send(message) to dispatch messages to a specific topic and tag.

@Component
public class Producer {
    @Value("${apache.rocketmq.producer.producerGroup}")
    private String producerGroup;

    @Value("${apache.rocketmq.namesrvAddr}")
    private String namesrvAddr;

    private DefaultMQProducer producer;

    @PostConstruct
    public void defaultMQProducer() {
        producer = new DefaultMQProducer(producerGroup);
        producer.setNamesrvAddr(namesrvAddr);
        producer.setVipChannelEnabled(false);
        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    public String send(String topic, String tags, String body) throws Exception {
        Message message = new Message(topic, tags, body.getBytes(RemotingHelper.DEFAULT_CHARSET));
        SendResult result = producer.send(message);
        return "{\"MsgId\":\"" + result.getMsgId() + "\"}";
    }
}

To consume messages, implement a DefaultMQPushConsumer. This is typically done by implementing CommandLineRunner to start the listener when the Spring Boot application launches.

Key configuration options:

• subscribe(topic, tag): Subscribes to a specific topic and tag.
• setConsumeFromWhere(ConsumeFromWhere): Determines where to start consuming (e.g., CONSUME_FROM_FIRST_OFFSET).
• setConsumeMessageBatchMaxSize(int): Sets how many messages to process in one batch.
• registerMessageListener(...): Defines the logic to execute when a message is received. Return ConsumeConcurrentlyStatus.CONSUME_SUCCESS to acknowledge receipt.

@Component
public class Consumer implements CommandLineRunner {
    @Value("${apache.rocketmq.consumer.pushConsumer}")
    private String pushConsumer;

    @Value("${apache.rocketmq.namesrvAddr}")
    private String namesrvAddr;

    public void messageListener() {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("SpringBootRocketMqGroup");
        consumer.setNamesrvAddr(namesrvAddr);
        try {
            consumer.subscribe("PushTopic", "push");
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            consumer.setConsumeMessageBatchMaxSize(1);

            consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
                for (Message msg : msgs) {
                    System.out.println("Received: " + new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
            consumer.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void run(String... args) throws Exception {
        this.messageListener();
    }
}