SpringBoot整合RabbitMQ_springboot整合mq

SpringBoot整合RabbitMQ

1.引入依赖

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

 2. 配置RabbitMQ连接

 在application.properties或application.yml中配置RabbitMQ服务器的连接参数

3. DirectExchange

3.1消费者

package com.jyx.consumer;
 
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;
 
 
@Configuration
 
public class DirectConsumer {
    // 声明队列
    @Bean
    public Queue queue(){
        return QueueBuilder.durable("Direct_01").build();
    }
    // 声明交换机
    @Bean
    public DirectExchange exchange(){
        return ExchangeBuilder.directExchange("Exchange_01").build();
    }
    // 绑定交换机和队列
    @Bean
    public Binding binding(Queue queue,DirectExchange exchange){
        return  BindingBuilder.bind(queue).to(exchange).with("RK01");
    }
    // 声明消费者
    @RabbitListener(queues = "Direct_01")
    public void receive(String msg){
        System.out.println("Direct_01消费者收到消息："+msg);
    }
 
}

 3.2 生产者

package com.jyx.producer;
 
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
 
@Service
public class DirectProducer {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    // 生产者 发送消息
    public void send(Object message) {
        rabbitTemplate.convertAndSend("Exchange_01", "RK01", message);
        System.out.println("发送消息：" + message);
    }
}

4. 启动类

package com.jyx;
 
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
 
@SpringBootApplication
public class App {
    public static void main(String[] args) {
        SpringApplication.run(App.class, args);
    }
}

  测试类

package com.jyx;
 
import com.jyx.model.OrderKO;
import com.jyx.producer.DirectProducer;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
 
import java.io.IOException;
 
@SpringBootTest
class AppTest {
    @Autowired
    private DirectProducer directProducer;
    @Test
    void contextLoads() throws IOException {
        for (int i = 0; i < 10; i++) {
        directProducer.send("hello world");
        }
        System.in.read();
    }
 
 
}

传对象

创建一个对象

package com.jyx.model;
 
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
 
import java.io.Serializable;
 
@Data
@AllArgsConstructor
@NoArgsConstructor
@Builder
public class OrderKO implements Serializable {
    private int id;
    private String name;
}

消费者

把String换成对象

 // 声明消费者
    @RabbitListener(queues = "Direct_01")
    public void receive(OrderKO msg){
        System.out.println("Direct_01消费者收到消息："+msg);
    }

测试类

 @Test
    void contextLoads2() throws IOException {
        for (int i = 0; i < 10; i++) {
            directProducer.send(OrderKO.builder().id(i).name("张"+i).build());
        }
        System.in.read();
    }

正常情况下

加工具类

package com.jyx.config;
 
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
@Configuration
public class RabbitConfig {
    @Bean
    public MessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }
}