Message Queue Fundamentals

A message queue (MQ) serves as a data structure for storing messages in a first-in-first-out manner.

Key Functions of MQ Systems

• Decoupling Applications: Enables independent development and deployment of services
• Rapid System Updates: Facilitates quick modifications without system-wide impacts
• Traffic Management: Handles request spikes by smoothing out load distribution

Advantages and Disadvantages

Disadvantages Include:

• Reduced system availability due to added complexity
• Increased operational overhead
• Challenges with asynchronous messaging:
  • Ensuring message ordering
  • Preventing message loss
  • Maintaining consistency
  • Managing duplicate deliveries

Popular Messaging Solutions

RocketMQ Overview

RocketMQ, an open-source middleware from Alibaba, evolved from MetaQ and was donated to Apache Foundation, becoming a top-level project within a year. It has been extensively used internally at Alibaba, handling massive loads during events like Singles' Day (2017 saw trillion-level message processing with peak TPS of 56 million).

Installation Instructions

Windows Setup

Refer to documentation for resolving "main class not found" errors during startup. Insure installation path contains no spaces.

Linux Environment

JDK Configuration

# Extract JDK
sudo tar -zxvf jdk-8u171-linux-x64.tar.gz

# Set environment variables
export JAVA_HOME=/opt/jdk1.8.0_171
export PATH=$PATH:${JAVA_HOME}/bin

# Reload configuration
source /etc/profile
java -version

Remove OpenJDK Conflicts

# Check installed Java packages
rpm -qa | grep java

# Remove conflicting packages
rpm -e --nodeps java-1.8.0-openjdk-1.8.0.232.b09-0.el7_7.x86_64

RocketMQ Setup

# Extract RocketMQ
unzip rocketmq-all-4.5.2-bin-release.zip
mv rocketmq-all-4.5.2-bin-release rocketmq

# Adjust memory settings in run scripts

Broker Configuration for Docker Compatibility

brokerClusterName = DefaultCluster
brokerName = broker-a
brokerId = 0
deleteWhen = 04
fileReservedTime = 48
brokerRole = ASYNC_MASTER
flushDiskType = ASYNC_FLUSH
brokerIP1=192.168.31.80
namesrvAddr=192.168.31.80:9876

Service Startup

# Start name server
sh mqnamesrv

# Start broker service
sh mqbroker -n localhost:9876 -c ../conf/broker.conf

# Disable firewall
systemctl stop firewalld.service

Testing

export NAMESRV_ADDR=localhost:9876
sh tools.sh org.apache.rocketmq.example.quickstart.Producer
sh tools.sh org.apache.rocketmq.example.quickstart.Consumer

Core Usage Patterns

Producer Implementation

public static void main(String[] args) throws Exception {
    DefaultMQProducer producer = new DefaultMQProducer("group1");
    producer.setNamesrvAddr("192.168.31.80:9876");
    producer.start();
    
    Message msg = new Message("topic1", "hello rocketmq".getBytes("UTF-8"));
    SendResult result = producer.send(msg);
    System.out.println("Result: " + result);
    
    producer.shutdown();
}

Consumer Implementation

public static void main(String[] args) throws Exception {
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
    consumer.setNamesrvAddr("192.168.31.80:9876");
    consumer.subscribe("topic1", "*");
    
    consumer.registerMessageListener((list, context) -> {
        for (MessageExt msg : list) {
            System.out.println("Received: " + new String(msg.getBody()));
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    });
    
    consumer.start();
    System.out.println("Consumer started");
}

Broadcasting Mode

consumer.setMessageModel(MessageModel.BROADCASTING);

Message Types

1. Synchronous: Immediate delivery with acknowledgment required
2. Asynchronous: Delivery with callback mechanism
3. One-way: No acknowledgment required

// Synchronous
Message syncMsg = new Message("topic2", "sync message".getBytes());
producer.send(syncMsg);

// Asynchronous
Message asyncMsg = new Message("topic2", "async message".getBytes());
producer.send(asyncMsg, new SendCallback() {
    public void onSuccess(SendResult result) { /* handle success */ }
    public void onException(Throwable e) { /* handle error */ }
});

// One-way
Message onewayMsg = new Message("topic2", "oneway message".getBytes());
producer.sendOneway(onewayMsg);

Delayed Messages

Message delayedMsg = new Message("topic3", "delayed message".getBytes());
delayedMsg.setDelayTimeLevel(3); // 30 seconds delay
producer.send(delayedMsg);

Batch Sending

List<Message> batchMessages = new ArrayList<>();
// Add messages to list
SendResult result = producer.send(batchMessages);

Message Filtering

Tag-based Filtering

// Producer
Message msg = new Message("topic6", "tag2", "filtered message".getBytes());

// Consumer
consumer.subscribe("topic6", "tag1 || tag2");

SQL-based Filtering

// Producer
msg.putUserProperty("vip", "1");
msg.putUserProperty("age", "20");

// Consumer
consumer.subscribe("topic7", MessageSelector.bySql("age >= 18"));

Ordered Messages

// Producer
SendResult result = producer.send(msg, new MessageQueueSelector() {
    public MessageQueue select(List<MessageQueue> queues, Message message, Object arg) {
        int index = order.getId().hashCode() % queues.size();
        return queues.get(index);
    }
}, null);

// Consumer
consumer.registerMessageListener(new MessageListenerOrderly() {
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
        for (MessageExt msg : msgs) {
            System.out.println("Processing: " + new String(msg.getBody()));
        }
        return ConsumeOrderlyStatus.SUCCESS;
    }
});

Transactional Messages

TransactionMQProducer producer = new TransactionMQProducer("group1");
producer.setTransactionListener(new TransactionListener() {
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        // Business logic here
        return LocalTransactionState.COMMIT_MESSAGE;
    }
    
    public LocalTransactionState checkLocalTransaction(MessageExt msgExt) {
        return LocalTransactionState.COMMIT_MESSAGE;
    }
});

Message transactionMsg = new Message("topic8", "transactional message".getBytes());
producer.sendMessageInTransaction(transactionMsg, null);

Configuration Requirements

Enable property filtering in broker configuration:

enablePropertyFilter=true

Start broker with updated config:

sh mqbroker -n localhost:9876 -c ../conf/broker.conf

Performance Considerations

• Message size limit: 4MB total per batch
• Batch message composition includes:
  • Topic string length
  • Body byte array
  • User properties (key-value pairs)
  • Log overhead (20 bytes fixed)