1.RabbitMQ依赖：如果是老版本(2.0.5)可能会不支持，尽量选择2.2以上
org.springframework.boot spring-boot-starter-parent2.5.0 org.springframework.boot spring-boot-starter-amqpcom.rabbitmq http-client3.12.0 org.codehaus.groovy groovyorg.codehaus.groovy groovy-json
2.配置文件
2.1 RabbitmqConfig(建立连接及管理等配置)
import com.rabbitmq.http.client.Client;
import com.rabbitmq.http.client.ClientParameters;
import com.rabbitmq.http.client.HttpComponentsRestTemplateConfigurator;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import java.net.MalformedURLException;
import java.net.URISyntaxException;
@Configuration
public class RabbitmqConfig {
@Bean
public MessageConverter messageConverter() {
return new Jackson2JsonMessageConverter("*");
}
private CachingConnectionFactory connectionFactory(String host, int port, String username,
String password, String virtualHost) {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
connectionFactory.setHost(host);
connectionFactory.setPort(port);
connectionFactory.setUsername(username);
connectionFactory.setPassword(password);
connectionFactory.setVirtualHost(virtualHost);
return connectionFactory;
}
@Bean(name = "opcUaConnectionFactory")
@Primary
public CachingConnectionFactory opcUaConnectionFactory(
@Value("${spring.rabbitmq.host}") String host,
@Value("${spring.rabbitmq.port}") int port,
@Value("${spring.rabbitmq.username}") String username,
@Value("${spring.rabbitmq.password}") String password,
@Value("${spring.rabbitmq.virtual-host}") String virtualHost) {
return connectionFactory(host, port, username, password, virtualHost);
}
@Bean(name = "opcUaRabbitAdmin")
@Primary
public RabbitAdmin opcUaRabbitAdmin(
@Qualifier("opcUaConnectionFactory") ConnectionFactory connectionFactory) {
RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
rabbitAdmin.setAutoStartup(true);
return rabbitAdmin;
}
@Bean(name = "opcUaRabbitListenerContainerFactory")
@Primary
public SimpleRabbitListenerContainerFactory opcUaRabbitListenerContainerFactory(
@Qualifier("opcUaConnectionFactory") ConnectionFactory connectionFactory) {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setConnectionFactory(connectionFactory);
factory.setMessageConverter(new Jackson2JsonMessageConverter());
return factory;
}
@Bean(name = "opcUaRabbitTemplate")
@Primary
public RabbitTemplate opcUaRabbitTemplate(@Qualifier("opcUaConnectionFactory") ConnectionFactory connectionFactory) {
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
rabbitTemplate.setMessageConverter(messageConverter());
return rabbitTemplate;
}
@Bean
public Client apiClient(@Value("${spring.rabbitmq.username}") String username,
@Value("${spring.rabbitmq.password}") String password,
@Value(("${spring.rabbitmq.web-api-url}")) String url) throws MalformedURLException, URISyntaxException {
return new Client(
new ClientParameters()
.url(url)
.username(username)
.password(password)
.restTemplateConfigurator(new HttpComponentsRestTemplateConfigurator())
);
}
}
2.2 RabbitQueueConfig(指定创建交换机、队列)
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Configuration;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
@Configuration
public class RabbitQueueConfig {
@Qualifier("opcUaRabbitAdmin")
@Resource
private RabbitAdmin rabbitAdmin;
@PostConstruct
public void init() {
TopicExchange exchange = new TopicExchange("opcua.topic", true, false);
rabbitAdmin.declareExchange(exchange);
Queue baseDataQueue = QueueBuilder.durable("opcua.data").build();
rabbitAdmin.declareQueue(baseDataQueue);
rabbitAdmin.declareBinding(BindingBuilder.bind(baseDataQueue).to(exchange).with("opcua.data"));
}
}
2.3 ThreadPoolConfig(配置线程池)
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
@Configuration
public class ThreadPoolConfig {
@Bean("opcUaDataExecutor")
public ThreadPoolExecutor taskExecutor(
@Value("${data-sync.thread-pool.blocking-queue-size}") int blockingQueueSize,
@Value("${data-sync.thread-pool.core-pool-size}") int corePoolSize,
@Value("${data-sync.thread-pool.maximum-pool-size}") int maximumPoolSize,
@Value("${data-sync.thread-pool.keep-alive-time}") long keepAliveTime) {
ArrayBlockingQueue queue = new ArrayBlockingQueue(blockingQueueSize);
ThreadFactory threadFactory = r -> new Thread(r, "data-sync");
return new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, TimeUnit.MILLISECONDS, queue, threadFactory);
}
}
2.4 OpcUaDataPayload 队列中数据对象定义
import lombok.Data;
import java.io.Serializable;
@Data
public class OpcUaDataPayload implements Serializable {
private static final long serialVersionUID = 1L;
private Integer eventType;
private String node;
private String value;
private String time;
public OpcUaDataPayload(Integer eventType, String node) {
this.eventType = eventType;
this.node = node;
}
public OpcUaDataPayload(Integer eventType, String node, String value, String time) {
this.eventType = eventType;
this.node = node;
this.value = value;
this.time = time;
}
public OpcUaDataPayload(Integer eventType, String node, String value) {
this.eventType = eventType;
this.node = node;
this.value = value;
}
public OpcUaDataPayload() {
}
}
2.5 配置文件变量值
#rabbitmq spring.rabbitmq.host=192.168.206.165 spring.rabbitmq.opcua.routing-key=opcua.data spring.rabbitmq.password=123456 spring.rabbitmq.port=5672 spring.rabbitmq.username=tg spring.rabbitmq.virtual-host=tg spring.rabbitmq.web-api-url=http://192.168.206.165:15672/api/ data-sync.mq-timeout=3600000 data-sync.thread-pool.blocking-queue-size=1000 data-sync.thread-pool.core-pool-size=50 data-sync.thread-pool.keep-alive-time=5000 data-sync.thread-pool.maximum-pool-size=1000
3 . OpcUaDataSender(发送消息到任务队列)
import com.templete.vo.OpcUaDataPayload;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
@Component
public class OpcUaDataSender {
@Resource
private RabbitTemplate rabbitTemplate;
@Value("${spring.rabbitmq.opcua.routing-key}")
private String routingKey;
public void send(OpcUaDataPayload payload) {
rabbitTemplate.convertAndSend(routingKey, payload);
}
}
4 . 任务处理(消费)
import com.rabbitmq.client.Channel;
import com.templete.vo.OpcUaDataPayload;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.util.concurrent.ThreadPoolExecutor;
@Component
@RabbitListener(
ackMode = "MANUAL",
containerFactory = "opcUaRabbitListenerContainerFactory",
queues = "opcua.data")
public class OpcUaDataReceiver {
private final Logger logger = LoggerFactory.getLogger(OpcUaDataReceiver.class);
// @Resource
// private EquipmentPartsService equipmentPartsService;
@Autowired
@Qualifier("opcUaDataExecutor")
private ThreadPoolExecutor executor;
@RabbitHandler
public void event(OpcUaDataPayload payload, Channel channel, Message message) {
executor.execute(() -> dispatch(payload, channel, message));
}
private void dispatch(OpcUaDataPayload opcUaDataPayload, Channel channel, Message message) {
boolean result = true;
logger.info("*******************info{}", message);
try {
switch (opcUaDataPayload.getEventType()) {
//操作能力
// case EventType.UPDATE:
// equipmentPartsService.updateByPoint(opcUaDataPayload);
// todo 操作数据 根据自己逻辑做操作
// break;
default:
}
} catch (Exception e) {
result = false;
logger.error(e.getMessage(), e);
} finally {
try {
// channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
if (result) {
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
} else {
channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
}
} catch (IOException e) {
logger.error(e.getMessage(), e);
}
}
}
}
