ActiveMQ
是一个高性能的消息服务, 它已经实现JMS
接口(Java消息服务(Java Message Service),Java平台中关于面向消息中间件的接口), 所以我们可以直接在 Java
中使用
使用场景: 多项目解耦合、分布式事务、流量控制等等
JMS
里面有一些概念, 我们提前说明一下JMS消息通常有两种类型:点对点(Point-to-Point)、发布/订阅(Publish/Subscribe)
cd bin/linux-x86-64/
./activemq start # 启动
启动之后可以看看图形界面版 localhots:8161
点击manager
, 默认账号密码是 admin
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-activemq</artifactId>
</dependency>
<!-- 连接池-->
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-pool</artifactId>
</dependency>
# 连接地址 默认连接端口为 61616
spring.activemq.broker-url=tcp://localhost:61616
spring.activemq.user=admin
spring.activemq.password=admin
# 开启连接池 最大连接数100
spring.activemq.pool.enabled=true
spring.activemq.pool.max-connections=100
package com.example.demo;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.web.servlet.MultipartConfigFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.jms.annotation.EnableJms;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.util.unit.DataSize;
import javax.jms.ConnectionFactory;
import javax.jms.Queue;
import javax.servlet.MultipartConfigElement;
@SpringBootApplication
@MapperScan("com.example.demo.mapper") // 直至mapper
@EnableJms // 开启jms
public class DemoApplication {
public static void main(String[] args) {
SpringApplication.run(DemoApplication.class, args);
}
// 配置默认队列
@Bean
public Queue queue(){
return new ActiveMQQueue("common.queue");
}
@Bean
public ConnectionFactory connectionFactory(){
System.out.println("创建连接");
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
connectionFactory.setBrokerURL("tcp://localhost:61616");
connectionFactory.setUserName("admin");
connectionFactory.setPassword("admin");
return connectionFactory;
}
@Bean
public JmsTemplate genJmsTemplate(){
return new JmsTemplate(connectionFactory());
}
@Bean
public JmsMessagingTemplate jmsMessagingTemplate() {
return new JmsMessagingTemplate(connectionFactory());
}
package com.example.demo.service;
import javax.jms.Destination;
/**
* 消息生产者
*/
public interface ProducerService {
/**
*指定消息队列 、消息
* @param destination
* @param message
*/
public void sendMessage(Destination destination, final String message);
/**
* 使用默认消息队列发送消息
* @param message
*/
public void sendMessage(final String message);
}
package com.example.demo.service.impl;
import com.example.demo.service.ProducerService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.stereotype.Service;
import javax.jms.Queue;
@Service
public class ProducerServiceImpl implements ProducerService {
@Autowired
private JmsMessagingTemplate jmsMessagingTemplate;
@Autowired
private Queue queue;
// 发送消息 destination是发送到队列 message是待发送的消息
@Override
public void sendMessage(javax.jms.Destination destination, String message) {
jmsMessagingTemplate.convertAndSend(destination, message);
}
@Override
public void sendMessage(String message) {
jmsMessagingTemplate.convertAndSend(this.queue, message);
}
}
package com.example.demo.jms;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;
@Component
public class OrderConsumer {
@JmsListener(destination = "order.queue")
public void receiveQueue(String text) {
System.out.println("接受到的消息 ====" + text);
}
}
package com.example.demo.controller;
import com.example.demo.bean.JsonUtils;
import com.example.demo.service.ProducerService;
import org.apache.activemq.command.ActiveMQQueue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import javax.jms.Destination;
@RestController
@RequestMapping("/api/v1/mq")
public class OrderController {
@Autowired
private ProducerService producerService;
@GetMapping("order")
public Object order(String msg) {
Destination destination = new ActiveMQQueue("order.queue");
producerService.sendMessage(destination, msg);
return JsonUtils.buildSuccess("ok");
}
@GetMapping("common")
public Object common(String msg) {
producerService.sendMessage(msg);
return JsonUtils.buildSuccess("ok");
}
}
由于common队列我们并没有去创建消费者, 所以消息会堆积,但是order对列是有消费者的, 来让我们看看情况
正常项目肯定生产者和消费者不在同一个项目里面, 这里只是简单演示一下。 主要步骤就是: 导入依赖 -> 配置mq -> controller -> 生产者 -> 消费者
本文为作者原创,手码不易,允许转载,转载后请以链接形式说明文章出处。