前言：只有消费者部分，没有记录生产者部分
结构图
配置类
可以xml配置，也可以配置类，二者可以相互转化。两种bean注入的方式。
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
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;@Configuration
public class DbDataQueueConsumeConfig {@Value("${rabbitmq.host}")private String host;@Value("${rabbitmq.port:5672}")
// @Value("#{new Integer('${rabbitmq.port}')}")private Integer port;@Value("${rabbitmq.username}")private String username;@Value("${rabbitmq.password}")private String password;@Value("${rabbitmq.virtual-host}")private String virtualHost;@Value("${rabbitmq.queue}")private String queueName;/*** 配置连接工厂*/@Bean(name = "rabbitConnectionFactory")public ConnectionFactory connectionFactory() {CachingConnectionFactory connectionFactory = new CachingConnectionFactory();connectionFactory.setHost(host);connectionFactory.setPort(port);connectionFactory.setUsername(username);connectionFactory.setPassword(password);connectionFactory.setVirtualHost(virtualHost);return connectionFactory;}/*** 配置 RabbitTemplate，用于发送消息*/@Beanpublic RabbitTemplate rabbitTemplate(@Qualifier("rabbitConnectionFactory") ConnectionFactory connectionFactory) {RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);rabbitTemplate.setMessageConverter(jsonMessageConverter());return rabbitTemplate;}@Beanpublic Jackson2JsonMessageConverter jsonMessageConverter() {return new Jackson2JsonMessageConverter();}/*** 配置消息监听容器*/@Beanpublic MessageListenerContainer messageListenerContainer(ConnectionFactory connectionFactory, BdDataQueueListener bdDataQueueListener) {SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();container.setConnectionFactory(connectionFactory);container.setQueueNames(queueName);container.setMessageListener(new MessageListenerAdapter(bdDataQueueListener));return container;}}配置文件中配置好rabbitmq服务的地址
rabbitmq.host=xxx.xxx.xxx.xxx
rabbitmq.port=3011
rabbitmq.username=username
rabbitmq.password=password
rabbitmq.virtual-host=/xxx
rabbitmq.queue=queuename
消费者
@Component
public class BdDataQueueListener implements MessageListener {@Overridepublic void onMessage(Message message) {try {// 这里写业务逻辑} catch (Exception e) {}}}重要部分都在上面。其余的不在放出来。
参考文章
文章1
文章2
