欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

springboot 整合 rabbitmq

程序员文章站 2022-03-07 17:05:01
...

    这篇文章,我们使用springboot来简单整合一下RabbitMQ,由于sprinboot的 “约定优于配置“,使得我们整合起来非常方便。参考文档

 

实现功能:

  1.  自动申明队列、交换器和绑定
  2.  使用自定义的MessageConvert如何进行配置
  3.  如何传递 JavaBean 消息和普通消息
  4.  如何手动签收消息

实现要点:

    ** 自定申明队列、交换器和绑定

          1、通过使用@Queue等注解实现

          2、使用@RabbitListener来实现

    ** 使用自定义的MessageConvert

          1、这个只需要 MessageConvert 申明的Bean 存在 @Bean注解且可以被Spring扫描到即可

    ** 传递 JavaBean 消息

          1、传递的 JavaBean 消息需要实现 java.io.Serializable 接口(默认情况下)

    ** 消费端如何监听消息

          1、使用@RabbitListener和@RabbitHandler结合使用

 

springboot整合rabbitmq自动配置位置:

springboot 整合 rabbitmq
            
    
    博客分类: springbootrabbitmq springbootrabbitmqspringboot整合rabbitmqrabbitmq整合sprinboot消息队列 

整合步骤:

1、引入amqp依赖

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
        <version>1.5.17.RELEASE</version>
</dependency>

2、rabbitmq 的springboot 配置

server:
  port: 9087
spring:
  rabbitmq:
    host: 140.143.237.224
    port: 5672
    username: root
    password: root
    virtual-host: /
    connection-timeout: 10000
    listener:
      simple:
        acknowledge-mode: manual # 手动应答
        auto-startup: true
        default-requeue-rejected: false # 不重回队列
        concurrency: 5
        max-concurrency: 20
        prefetch: 1 # 每次只处理一个信息
        retry:
          enabled: false

3、配置自定义的消息转换器

/**
 * rabbitmq 配置类
 *
 * @author huan.fu
 * @date 2018/10/24 - 19:42
 */
@Configuration
@Slf4j
public class RabbitConfiguration {

	@Bean
	public MessageConverter messageConverter() {
		return new SimpleMessageConverter() {
			@Override
			protected Message createMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
				Message message = super.createMessage(object, messageProperties);
				log.info("使用自定义的MessageConvert转换消息");
				return message;
			}
		};
	}
}

4、编写用户实体类

/**
 * 用户实体类
 *
 * @author huan.fu
 * @date 2018/10/22 - 15:35
 */
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class User implements Serializable {
	private Integer userId;
	private String username;
	private String password;
}

    注意:这个实体类需要实现序列化接口

5、创建消息生产者

@Component
public class RabbitProducer {

	@Autowired
	private AmqpTemplate amqpTemplate;

	/**
	 * 发送消息
	 */
	@SendTo
	public void sendMessage() {
		new Thread(() -> {
			try {
				TimeUnit.SECONDS.sleep(3);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			// 发送简单消息
			IntStream.rangeClosed(1, 10).forEach(num -> {
				String body = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()) + " : " + num;
				MessageProperties properties = new MessageProperties();
				properties.setContentEncoding("UTF-8");
				properties.setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
				Message message = new Message(body.getBytes(Charset.forName(properties.getContentEncoding())), properties);
				amqpTemplate.convertAndSend("rabbit-springboot-exchange", "rabbit-springboot-routingkey", message);
			});
			// 发送java bean 消息
			IntStream.rangeClosed(1, 10).forEach(num -> {
				User user = User.builder().userId(num).username("zhangsan:" + num).password("666666").build();
				amqpTemplate.convertAndSend("rabbit-springboot-exchange", "rabbit-springboot-routingkey.javabean", user);
			});
		}).start();
	}
}

    注意: AmqpTemplate 为springboot 默认配置的

6、创建消息接收者

public class RabbitReceiver {

	@RabbitHandler
	@RabbitListener(bindings = @QueueBinding(
			value = @Queue(value = "rabbit-springboot-queue", durable = "true", exclusive = "false", autoDelete = "false"),
			exchange = @Exchange(value = "rabbit-springboot-exchange", type = ExchangeTypes.DIRECT, durable = "true", autoDelete = "false"),
			key = "rabbit-springboot-routingkey"
	))
	public void receiveMessage(Message message, Channel channel) {
		String encoding = message.getMessageProperties().getContentEncoding();
		log.info("接收到消息1:[{}]", new String(message.getBody(), Charsets.toCharset(encoding)));
		try {
			channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
		} catch (IOException e) {
			log.error(e.getMessage(), e);
		}
	}

	@RabbitHandler
	@RabbitListener(bindings = @QueueBinding(
			value = @Queue(value = "rabbit-springboot-queue-javabean", durable = "true", exclusive = "false", autoDelete = "false"),
			exchange = @Exchange(value = "rabbit-springboot-exchange", type = ExchangeTypes.DIRECT, durable = "true", autoDelete = "false"),
			key = "rabbit-springboot-routingkey.javabean"
	))
	public void receiveMessage(User user, Message message, Channel channel) {
		log.info("接收到消息2:[{}]", user);
		try {
			channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
		} catch (IOException e) {
			log.error(e.getMessage(), e);
		}
	}
}

    注意:此处使用 @RabbitListener 来自动创建队列、交换器和绑定等

7、常见消息启动类

@SpringBootApplication
@Slf4j
public class Application implements ApplicationRunner {

	@Autowired
	private RabbitProducer rabbitProducer;

	public static void main(String[] args) {
		SpringApplication.run(Application.class, args);
	}

	@Override
	public void run(ApplicationArguments applicationArguments) throws Exception {
		rabbitProducer.sendMessage();
	}
}

 8、运行结果springboot 整合 rabbitmq
            
    
    博客分类: springbootrabbitmq springbootrabbitmqspringboot整合rabbitmqrabbitmq整合sprinboot消息队列 
 

完成代码:

springboot整合rabbitmq:https://gitee.com/huan1993/rabbitmq/tree/master/rabbitmq-springboot

   

  • springboot 整合 rabbitmq
            
    
    博客分类: springbootrabbitmq springbootrabbitmqspringboot整合rabbitmqrabbitmq整合sprinboot消息队列 
  • 大小: 91.5 KB
  • springboot 整合 rabbitmq
            
    
    博客分类: springbootrabbitmq springbootrabbitmqspringboot整合rabbitmqrabbitmq整合sprinboot消息队列 
  • 大小: 27.1 KB