Background Jobs & Message Queues
Relevant source files
The following files were used as context for generating this wiki page:
- zheng-api/zheng-api-server/src/main/java/com/zheng/api/server/jms/DefaultMessageQueueListener.java
- zheng-api/zheng-api-server/src/main/resources/applicationContext-activemq.xml
- zheng-cms/zheng-cms-job/src/main/java/com/zheng/cms/job/jms/DefaultMessageQueueListener.java
- zheng-cms/zheng-cms-web/pom.xml
- zheng-message/zheng-message-server/pom.xml
- zheng-cms/zheng-cms-job/pom.xml
The zheng project utilizes a variety of technologies for handling background jobs and asynchronous messaging, enabling different modules to perform tasks without blocking main application threads. This distributed approach enhances scalability and decouples system components. The primary technologies identified are ActiveMQ for JMS-based messaging, RocketMQ, tbschedule for distributed job scheduling, and netty-socketio for real-time communication.
These systems are primarily integrated within the zheng-api-server, zheng-cms-job, zheng-cms-web, and zheng-message-server modules. Each technology serves a different purpose, from traditional message queuing to complex distributed task scheduling and real-time event handling.
Messaging & Job Technologies Overview
The project employs several distinct libraries for asynchronous processing, as evidenced by the dependencies in various module pom.xml files.
| Technology | Module | Purpose | Source |
|---|---|---|---|
| ActiveMQ | zheng-api-server | Standard JMS message queuing. | applicationContext-activemq.xml |
| JMS Listener | zheng-api-server, zheng-cms-job | Consumes messages from a queue. | DefaultMessageQueueListener.java |
| tbschedule | zheng-cms-web | Distributed job scheduling framework. | zheng-cms-web/pom.xml:35-41 |
| RocketMQ | zheng-cms-web | Distributed message queue. | zheng-cms-web/pom.xml:43-47 |
| netty-socketio | zheng-message-server | Real-time, event-based communication. | zheng-message-server/pom.xml:23-29 |
ActiveMQ Integration (zheng-api-server)
The zheng-api-server module is configured to use ActiveMQ for asynchronous message processing. The integration is defined through Spring Framework's JMS support.
Sources: zheng-api/zheng-api-server/src/main/resources/applicationContext-activemq.xml, zheng-api/zheng-api-server/src/main/java/com/zheng/api/server/jms/DefaultMessageQueueListener.java
Configuration
The core of the ActiveMQ integration is in applicationContext-activemq.xml. This file defines the necessary Spring beans to connect to an ActiveMQ broker, send messages, and listen for incoming messages on a specific queue.
Key Configuration Beans
| Bean ID / Type | Description | Source |
|---|---|---|
targetConnectionFactory | The raw ActiveMQ connection factory, configured with the broker URL. | applicationContext-activemq.xml:13-17 |
connectionFactory | A Spring-managed SingleConnectionFactory that wraps the target factory for efficient connection management. | applicationContext-activemq.xml:19-23 |
jmsTemplate | A Spring helper class that simplifies sending JMS messages. | applicationContext-activemq.xml:25-28 |
queueDestination | Defines the specific JMS queue to interact with. The queue name is zheng-api-queue. | applicationContext-activemq.xml:30-33 |
jmsContainer | A listener container that binds a message listener to a destination, automatically receiving messages. | applicationContext-activemq.xml:36-41 |
The broker connection is configured with a property placeholder ${activemq.brokerURL}.
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="${activemq.brokerURL}"/>
</bean>
Sources: zheng-api/zheng-api-server/src/main/resources/applicationContext-activemq.xml:14-16
The listener container connects the messageQueueListener bean to the zheng-api-queue.
<jms:listener-container container-type="default" connection-factory="connectionFactory" acknowledge="auto">
<jms:listener destination="zheng-api-queue" ref="messageQueueListener"/>
</jms:listener-container>
Sources: zheng-api/zheng-api-server/src/main/resources/applicationContext-activemq.xml:36-38
Message Listener
The DefaultMessageQueueListener class is responsible for processing messages received from the queue. It implements the standard JMS MessageListener interface.
This diagram illustrates the class structure.
The onMessage method is the entry point for message handling. In its current implementation, it simply logs the content of the received message, indicating a basic setup for asynchronous task execution.
public class DefaultMessageQueueListener implements MessageListener {
private static Logger _log = LoggerFactory.getLogger(DefaultMessageQueueListener.class);
public void onMessage(Message message) {
try {
_log.info("onMessage: {}", message);
} catch (Exception e) {
e.printStackTrace();
}
}
}
Architecture and Message Flow
The following diagram shows the flow of a message from a producer to the listener within the zheng-api-server.
CMS Module Integration
The zheng-cms parent module also contains submodules that leverage messaging and job scheduling.
JMS Listener (zheng-cms-job)
The zheng-cms-job module contains a DefaultMessageQueueListener which is identical in implementation to the one in zheng-api-server. This suggests a common pattern for creating simple, queue-based background workers across the project. The primary purpose of this module is to run background jobs, and this listener is a component of that system.
Sources: zheng-cms/zheng-cms-job/src/main/java/com/zheng/cms/job/jms/DefaultMessageQueueListener.java, zheng-cms/zheng-cms-job/pom.xml
Other Technologies (zheng-cms-web)
The zheng-cms-web module's pom.xml reveals dependencies on two other significant frameworks for background processing:
-
tbschedule: A distributed scheduling framework from Taobao (Alibaba). This is likely used for more complex, coordinated jobs that need to run across multiple nodes. The dependency is included as a system-scoped JAR.
<dependency> <groupId>com.taobao.pamirs.schedule</groupId> <artifactId>tbschedule</artifactId> <version>3.2.8-SNAPSHOT</version> <scope>system</scope> <systemPath>${project.basedir}/src/main/webapp/WEB-INF/lib/tbschedule-3.2.8-SNAPSHOT.jar</systemPath> </dependency> -
RocketMQ: Another distributed messaging and streaming platform from Alibaba. Its inclusion suggests that for some use cases, the project may require a more feature-rich or higher-performance message queue than ActiveMQ.
<dependency> <groupId>com.alibaba.rocketmq</groupId> <artifactId>rocketmq-client</artifactId> <version>3.6.2.Final</version> </dependency>
Real-time Messaging (zheng-message-server)
Distinct from the asynchronous job queues, the zheng-message-server module is dedicated to real-time communication. It uses netty-socketio, a Java implementation of the Socket.IO protocol built on Netty. This is intended for features requiring persistent, bidirectional connections between the server and clients, such as live notifications or chat.
The dependency is included as a system-scoped JAR.
<dependency>
<groupId>com.corundumstudio.socketio</groupId>
<artifactId>netty-socketio</artifactId>
<version>1.7.12.zheng</version>
<scope>system</scope>
<systemPath>${basedir}/lib/netty-socketio-1.7.12.zheng.jar</systemPath>
</dependency>