Springboot RabbitMq 整合使用简单队列

喜欢ヅ旅行 2024-03-27 15:42 31阅读 0赞

### Springboot+RabbitMq整合使用 简单队列 ###

本项目主要有两个服务，分别是是consumer和producer。

### consumer服务 ###

全部依赖

<dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency> 
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-amqp</artifactId>
            </dependency>
    
            <dependency>
                <groupId>com.google.guava</groupId>
                <artifactId>guava</artifactId>
                <version>19.0</version>
            </dependency>
            <dependency>
                <groupId>org.apache.commons</groupId>
                <artifactId>commons-lang3</artifactId>
                <version>3.4</version>
            </dependency>
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <version>RELEASE</version>
                <scope>compile</scope>
            </dependency>
            <dependency>
                <groupId>com.alibaba</groupId>
                <artifactId>fastjson</artifactId>
                <version>1.1.37</version>
            </dependency>

配置文件

server:
      port: 8081
    spring:
      rabbitmq:
        host: 127.0.0.1
        port: 5672
        username: guest #用户名
        password: guest  #密码
        virtual-host:

实体

import lombok.AllArgsConstructor;
    import lombok.Data;
    import lombok.NoArgsConstructor; 
    import java.io.Serializable;
    
    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public class User implements Serializable {
        
        // 名字
        private String name;
        // 年龄
        private Integer age;
        // 性别
        private String sex; 
    }

接收者

import com.alibaba.fastjson.JSON;
    import com.lideru.consumer.pojo.User; 
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener; 
    import org.springframework.stereotype.Component; 
    import java.util.Date;
    @Component
    @RabbitListener(queues = "HelloWorld")
    public class HelloWorldReceiver {
        
    
    
        @RabbitHandler
        public void process(String user ) {
        
            
            System.out.println( new Date() +" 接收 : " + user );
        }
    }

配置类

import org.springframework.amqp.core.Queue;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration; 
    @Configuration
    public class RabbitHelloWorldConfig {
         
        /**
         * Queue 可以有4个参数
         *      1.name          队列名
         *      2.durable       持久化消息队列 ,rabbitmq重启的时候不需要创建新的队列 默认true
         *      3.auto-delete   表示消息队列没有在使用时将被自动删除 默认是false
         *      4.exclusive     表示该消息队列是否只在当前connection生效,默认是false
         */
        @Bean
        public Queue createHelloWorldQueue() {
         
            return new Queue("HelloWorld",true);
        } 
    }

### producer服务 ###

全部依赖

配置文件

server:
      port: 8080
    spring:
      rabbitmq:
        host: 127.0.0.1
        port: 5672
        username: guest #用户名
        password: guest  #密码
        virtual-host:

实体

发送者

import org.springframework.amqp.rabbit.connection.CorrelationData;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    import java.util.Date;
    import java.util.UUID; 
    @Component
    public class HelloWorldSender {
         
    	@Autowired
    	private RabbitTemplate rabbitTemplate;
    
    	public void send() {
        
    		CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
    		String content = "HelloWorld = " + new Date() + ", content = " + UUID.randomUUID().toString();
    		this.rabbitTemplate.convertAndSend("HelloWorld",(Object) content,correlationData);
    		System.out.println("发送成功,"+new Date()+","+content);
    	}  
    }

测试

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;
    
    @RestController
    @RequestMapping
    public class HelloWorldTestController {
        
        @Autowired
        private HelloWorldSender helloWorldSender; 
        @GetMapping("/helloWorld")
        public String hello() throws InterruptedException {
        
            helloWorldSender.send();
            return "Hello!";
        }
    }

服请求 http://localhost:8080/helloWorld  
producer控制台

Fri Feb 10 16:02:42 CST 2023发送成功,User(name=HelloWorld, age=18, sex=man)

consumer控制台

Fri Feb 10 16:02:42 CST 2023 接收 : User(name=HelloWorld, age=18, sex=man)

到这里就实现了，简简单单。  
  
  
我们先关闭consumer服务，在请求http://localhost:8080/helloWorld，然后再启动consumer服务。结果如下  
producer控制台

Fri Feb 10 16:04:14 CST 2023发送成功,User(name=HelloWorld, age=18, sex=man)

consumer控制台

Fri Feb 10 16:04:22 CST 2023 接收 : User(name=HelloWorld, age=18, sex=man)

即使服务挂了，队列还在，等服务再启动起了时就自动跑队列里面的东西咯。