Springboot项目整合Rabbitmq详细教程

「爱情、让人受尽委屈。」 2022-09-15 14:50 187阅读 0赞

首先在springboot的pom文件里引入依赖

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

然后application.yml：

ps：里面的虚拟host配置项不是必须的，我自己在rabbitmq服务上创建了自己的虚拟host，所以我配置了；你们不创建，就不用加这个配置项。

spring:
      application:
        name: 3d-gis
      datasource:
        driver-class-name: org.postgresql.Driver
        url: jdbc:postgresql://172.16.10.201:5432/3d_gis
        username: postgres
        password: postgres
      rabbitmq:
        publisher-confirms: true
        publisher-returns: true
        host: 172.16.10.201
        port: 5672
        username: leaniot
        password: leaniot
        virtual-host: /3d_gis
        listener:
          simple:
            acknowledge-mode: manual
          direct:
            acknowledge-mode: manual

RabbitMQ配置类

import org.springframework.amqp.core.*;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    
    
    /**
     * RabbitMQ 配置
     *
     * @author tarzan
     * @version 1.0
     * @date 2021年05月18日 09:15:40
     * @since JDK1.8
     */
    
    @Configuration
    public class RabbitMQConfig{
    
        /** gis图形交换机 */
        public static final String  GIS_GRAPHICS_EXCHANGE = "3d_gis_exchange";
        /** gis图形数据接收序列 */
        public static final String GIS_DATA_RECEIVE_REQUEUE = "gis_data_queue";
        /** gis图形数据返回（发送）序列 */
        public static final String GIS_DATA_SEND_QUEUE = "gis_result_queue";
    
        @Bean
        public Queue gisDataReceiveQueue () {
            return new Queue(GIS_DATA_RECEIVE_REQUEUE);
        }
    
        @Bean
        public Queue gisDataSendQueue () {
            return new Queue(GIS_DATA_SEND_QUEUE);
        }
    
    
        @Bean
        public DirectExchange directExchange() {
            return new DirectExchange(GIS_GRAPHICS_EXCHANGE);
        }
    
        @Bean
        public Binding receiveBinding () {
            return BindingBuilder.bind(gisDataReceiveQueue()).to(directExchange()).withQueueName();
        }
    
        @Bean
        public Binding sendBinding () {
            return BindingBuilder.bind(gisDataSendQueue()).to(directExchange()).withQueueName();
        }
    
    
    
    }

RabbitSender类

import java.util.UUID;
    
    import lombok.extern.slf4j.Slf4j;
    import org.springblade.gis.rabbitmq.confirm.RabbitReturnCallback;
    import org.springblade.gis.rabbitmq.confirm.RabbitConfirmCallback;
    import org.springframework.amqp.rabbit.connection.CorrelationData;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.InitializingBean;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.Resource;
    
    @Component
    @Slf4j
    public class RabbitSender implements InitializingBean {
    
        @Resource
        private RabbitTemplate rabbitTemplate;
    
        /**
         * 	对外发送消息的方法
         * @param message 	具体的消息内容
         * @throws Exception
         */
        public void send(Object message)  {
            CorrelationData data = new CorrelationData(UUID.randomUUID().toString());
            rabbitTemplate.convertAndSend(RabbitMQConfig.GIS_GRAPHICS_EXCHANGE, RabbitMQConfig.GIS_DATA_SEND_QUEUE, message, data);
        }
    
        @Override
        public void afterPropertiesSet () {
            rabbitTemplate.setConfirmCallback(new RabbitConfirmCallback());
            rabbitTemplate.setReturnCallback(new RabbitReturnCallback());
            rabbitTemplate.setMandatory(true);
        }
    }

RabbitReceiver类

import com.alibaba.fastjson.JSONArray;
    import com.alibaba.fastjson.JSONObject;
    import lombok.extern.slf4j.Slf4j;
    import org.springblade.gis.modules.db.service.DbDataService;
    import org.springframework.amqp.rabbit.annotation.*;
    import org.springframework.amqp.core.Message;
    import org.springframework.stereotype.Component;
    import com.rabbitmq.client.Channel;
    
    import javax.annotation.Resource;
    import java.io.IOException;
    import java.util.ArrayList;
    import java.util.List;
    import java.util.Map;
    
    @Component
    @Slf4j
    public class RabbitReceiver {
    
        @Resource
        private DbDataService dbDataService;
        @Resource
        private RabbitSender sender;
    
    
    
        /**
         * 	组合使用监听
         * @param message
         * @param channel
         */
        @RabbitListener(queues =RabbitMQConfig.GIS_DATA_RECEIVE_REQUEUE)
        public void onMessage(String msg,Message message, Channel channel) {
            //	1. 收到消息以后进行业务端消费处理
            String body=new String(message.getBody());
            log.info("消费消息:" + body );
            JSONObject json;
            try {
                json=JSONObject.parseObject(body);
            } catch (Exception e) {
                ackOrReject(message,channel,true);
                return;
            }
            String table= json.getString("table");
            JSONArray data= json.getJSONArray("data");
            List<Map<String,Object>> maps=new ArrayList<>();
            data.forEach(e->{
               Map<String,Object> map=(Map<String, Object>) e;
                maps.add(map);
             });
             List<Long> ids=dbDataService.saveBatch(table,maps);
             sender.send(JSONArray.toJSONString(ids));
            //  2. 处理成功之后 获取deliveryTag 并进行手工的ACK操作, 因为我们配置文件里配置的是 手工签收
             ackOrReject(message,channel,true);
        }
    
        private void ackOrReject(Message message, Channel channel, boolean result)  {
            try {
                if (result) {
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                } else {
                    channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
                }
            }catch (IOException e){
                new IOException();
            }
        }
    
    
    }

测试类

import io.swagger.annotations.Api;
    import io.swagger.annotations.ApiOperation;
    import lombok.AllArgsConstructor;
    import org.springblade.core.secure.annotation.NoToken;
    import org.springblade.core.tool.api.R;
    import org.springblade.three.dimension.gis.rabbitmq.RabbitSender;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    import javax.annotation.Resource;
    
    @NoToken
    @RestController
    @AllArgsConstructor
    @RequestMapping("rabbit")
    @Api(value = "消息队列", tags = "消息队列")
    public class RabbitController {
        @Resource
        private RabbitSender sender;
    
    
        @ApiOperation(value = "添加 @author Tarzan Liu")
        @GetMapping("test")
        public R test(){
            sender.send("测试666");
            return R.status(true);
        }
    }

测试结果

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5rSb6Ziz5rOw5bGx_size_20_color_FFFFFF_t_70_g_se_x_16][]

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5rSb6Ziz5rOw5bGx_size_20_color_FFFFFF_t_70_g_se_x_16 1][]

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5rSb6Ziz5rOw5bGx_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/b1264167fa474f04ae2c5363e7d27948.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5rSb6Ziz5rOw5bGx_size_20_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220828/c94299f2d8644ba28c2b8333d1e887b3.png