solon 集成 activemq-client (sdk)

原始状态的 activemq-client sdk 集成非常方便，也更适合定制。就是有些同学，可能对原始接口会比较陌生，会希望有个具体的示例。

<dependency>     <groupId>org.apache.activemq</groupId>     <artifactId>activemq-client</artifactId>     <version>${activemq.version}</version> </dependency>  <dependency>     <groupId>org.apache.activemq</groupId>     <artifactId>activemq-pool</artifactId>     <version>${activemq.version}</version> </dependency> 

希望更加简化使用的同学，可以使用：

activemq-solon-cloud-plugin （使用更简单，定制性弱些）

1、添加集成配置

先使用 Solon 初始器 先生成一个 Solon Web 模板项目，然后添加上面的 activemq-client 依赖。再做个配置约定（也可按需定义）：

• "solon.activemq"，作为配置前缀
  • "properties"，作为公共配置
  • "producer"，作为生态者专属配置（估计用不到）
  • "consumer"，作为消费者专属配置（估计用不到）

具体的配置属性，参考自：ActiveMQConnectionFactory

solon.app:   name: "demo-app"   group: "demo"  # 配置可以自由定义，与 @Bean 代码对应起来即可（以下为参考） solon.activemq:   properties:  #公共配置（配置项，参考：ActiveMQConnectionFactory）     brokerURL: "failover:tcp://localhost:61616"     redeliveryPolicy:       initialRedeliveryDelay: 5000       backOffMultiplier: 2       useExponentialBackOff: true       maximumRedeliveries: -1       maximumRedeliveryDelay: 3600_000 

添加 java 配置器

@Configuration public class ActivemqConfig {     @Bean(destroyMethod = "stop")     public Connection client(@Inject("${solon.activemq.properties}") Props common) throws Exception {         String brokerURL = (String) common.remove("brokerURL");         String userName = (String) common.remove("userName");         String password = (String) common.remove("password");          ActiveMQConnectionFactory factory;         if (Utils.isEmpty(userName)) {             factory = new ActiveMQConnectionFactory(brokerURL);         } else {             factory = new ActiveMQConnectionFactory(brokerURL, userName, password);         }          //绑定额外的配置并创建连接         Connection connection = common.bindTo(factory).createConnection();         connection.start();         return connection;     }      @Bean     public IProducer producer(Connection connection) throws Exception {         return new IProducer(connection);     }      @Bean     public void consumer(Connection connection,                          MessageListener messageListener) throws Exception {         Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);          Destination destination = session.createTopic("topic.test");         MessageConsumer consumer = session.createConsumer(destination);          consumer.setMessageListener(messageListener);     } } 

activemq 的消息发送的代码比较复杂，所以我们可以做个包装处理（用于上面的配置构建），临时命名为 IProducer：

public class IProducer {     private Connection connection;      public IProducer(Connection connection) {         this.connection = connection;     }      public void send(String topic, MessageBuilder messageBuilder) throws JMSException {         Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);          Destination destination = session.createTopic(topic);         MessageProducer producer = session.createProducer(destination);          producer.send(destination, messageBuilder.build(session));     }      @FunctionalInterface     public static interface MessageBuilder {         Message build(Session session) throws JMSException;     } } 

3、代码应用

发送（或生产），这里代控制器由用户请求再发送消息（仅供参考）：

@Controller public class DemoController {     @Inject     private IProducer producer;      @Mapping("/send")     public void send(String msg) throws Exception {         //发送         producer.send("topic.test", s -> s.createTextMessage("test"));     } } 

监听（或消费），这里采用订阅回调的方式：（仅供参考）

@Component public class DemoMessageListener implements MessageListener {     @Override     public void onMessage(Message message) {         System.out.println(message);         RunUtil.runAndTry(message::acknowledge);     } }