首页 > 编程语言 >java——spring boot集成RabbitMQ——如何实现手动ack——topic——路由模式——代码展示

java——spring boot集成RabbitMQ——如何实现手动ack——topic——路由模式——代码展示

时间:2023-02-26 13:34:22浏览次数:42  
标签:java amqp spring boot springframework org import public

发送方:

pom文件:

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>org.example</groupId>
    <artifactId>springrmqtopicsender</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.4.5</version>
        <relativePath/>
    </parent>


    <dependencies>

        <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>org.springframework.boot</groupId>
            <artifactId>spring-boot-devtools</artifactId>
            <scope>runtime</scope>
            <optional>true</optional>
        </dependency>

    </dependencies>

</project>

 

 

 

 

 

 

 

yml配置:

spring:
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest

 

 

 

 

 

 

 

config配置:

package org.example.config;

import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * 主题交换机
 * topic策略可以根据routingKey的规则(通配符方式)进行去匹配队列进行转发规则为*.#.*
 */
@Configuration
public class RabbitTopicConfig
{


    public final static String TOPIC_NAME = "amqp-topic";


    @Bean
    TopicExchange topicExchange()
    {

        return new TopicExchange(TOPIC_NAME,true,false);

    }


}

 

 

 

 

 

 

 

 

发送代码:

package org.example.sender;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * 消息生产者 发送消息
 */
@Component
public class MessageSender {

    @Autowired
    RabbitTemplate rabbitTemplate;

    /**
     * 发送消息
     * @param info
     */
    public void send(String info)
    {

        System.out.println("发送消息>>>"+info);


        rabbitTemplate.convertAndSend("amqp-topic","huawei.a",info);
    }

}

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

启动服务发送:

package org.example.controller;

import org.example.sender.MessageSender;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @Auther: [email protected]
 * @Date: 2020/10/4 11:34
 */
@RestController
public class IndexController {

    @Autowired
    MessageSender messageSender;

    @RequestMapping("/index")
    public String index()
    {

        messageSender.send("中国——路由——华为");
        return "SUCCESS";
    }

}

 

 

 

 

 

 

 

 

 

package org.example;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class HelloWorldApplication
{

    public static void main(String[] args)
    {

        SpringApplication.run(HelloWorldApplication.class, args);

    }
}

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

==================================================================

 

 

 

 

 

 

 

 

 

手动ack:

 

pom文件:

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>org.example</groupId>
    <artifactId>springrmqtopicreceiver</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>


    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.4.5</version>
        <relativePath/>
    </parent>


    <dependencies>

        <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>org.springframework.boot</groupId>
            <artifactId>spring-boot-devtools</artifactId>
            <scope>runtime</scope>
            <optional>true</optional>
        </dependency>

    </dependencies>

</project>

 

 

 

 

 

 

 

 

yml文件:

spring:
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest
    #关闭自动ack,设置为手动ack
    listener:
      simple:
        acknowledge-mode: manual

 

 

 

 

 

 

 

 

配置文件:

package org.example.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * 主题交换机
 * topic策略可以根据routingKey的规则(通配符方式)进行去匹配队列进行转发规则为*.#.*
 */
@Configuration
public class RabbitTopicConfig {

    public final static String TOPIC_NAME = "amqp-topic";

    @Bean
    TopicExchange topicExchange(){
        return new TopicExchange(TOPIC_NAME,true,false);
    }

    @Bean
    Queue xiaomi(){
        return new Queue("xiaomi");
    }

    @Bean
    Queue huawei(){
        return new Queue("huawei");
    }

    @Bean
    Binding xiaomiBinding(){
        //xiaomi.#:表示消息的routingKey是以xiaomi开头的就会路由到xiaomi的队列
        return BindingBuilder.bind(xiaomi()).to(topicExchange()).with("xiaomi.#");
    }

    @Bean
    Binding huaweiBinding(){
        return BindingBuilder.bind(huawei()).to(topicExchange()).with("huawei.#");
    }
}

 

 

 

 

 

 

 

接收信息,并手动签收:

package org.example.receiver;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.io.IOException;

/**
 */
@Component
public class TopicReceiver {

    //分别监听名称为xiaomi、huawei的队列
    @RabbitListener(queues = "xiaomi")
    public void handlerXM(Message message,String msg, Channel channel) throws IOException {

        System.out.println("小米:"+msg);

        //手动签收,不启动批量签收
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
    }


    @RabbitListener(queues = "huawei")
    public void handlerHW(Message message,String msg, Channel channel) throws IOException {


        System.out.println("华为:"+msg);

        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
    }


}

 

 

 

 

 

 

 

 

 

 

启动服务:

package org.example;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class HelloWorldApplication
{

    public static void main(String[] args)
    {

        SpringApplication.run(HelloWorldApplication.class, args);

    }
}

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

标签:java,amqp,spring,boot,springframework,org,import,public
From: https://www.cnblogs.com/xiaobaibailongma/p/17156542.html

相关文章

  • 《JavaScript百炼成仙》,简单但是挺有效的
    编程之修,重在积累,而非资质。资质虽然重要,可是后天的努力更不可少。《JavaScript百炼成仙》是一本以玄幻小说的形式,来讲述JavaScript的知识。此篇仅仅是我快速阅读《JavaScri......
  • 【Java】“com.alibaba.fastjson.JSONObject cannot be cast to“报错问题
    【Java】修复"com.alibaba.fastjson.JSONObjectcannotbecastto"报错问题报错如下:java.lang.ClassCastException:com.alibaba.fastjson.JSONObjectcannotbecasttoc......
  • 14_SpringMVC_其他注解
     SpringMVC中注解完善1、@PostMapping作用:指定当前发送请求的方式只可以是post请求属性:  和@RequestMapping中属性一致代码实现@PostMapping("/userControllerA")pub......
  • 14_SpringMVC_其他注解
     SpringMVC中注解完善1、@PostMapping作用:指定当前发送请求的方式只可以是post请求属性:  和@RequestMapping中属性一致代码实现@PostMapping("/userControllerA")pub......
  • 13_SpringMVC_异常处理
    ​ SpringMVC异常简介系统中异常包括两类:预期异常(检查型异常)和运行时异常RuntimeException,前者通过捕获异常从而获取异常信息,后者主要通过规范代码开发、测试通过手......
  • 13_SpringMVC_异常处理
    ​ SpringMVC异常简介系统中异常包括两类:预期异常(检查型异常)和运行时异常RuntimeException,前者通过捕获异常从而获取异常信息,后者主要通过规范代码开发、测试通过手......
  • 14_SpringMVC_其他注解
    ​ SpringMVC中注解完善1、@PostMapping作用:指定当前发送请求的方式只可以是post请求属性:  和@RequestMapping中属性一致代码实现@PostMapping("/userContro......
  • 14_SpringMVC_其他注解
    ​ SpringMVC中注解完善1、@PostMapping作用:指定当前发送请求的方式只可以是post请求属性:  和@RequestMapping中属性一致代码实现@PostMapping("/userContro......
  • springMvc_快速入门
    概念:是一种基于Java实现mvc模型的轻量级web框架优点:使用简单,开发便捷   灵活性强总体来说springMvc就是来替代servlet的一种工具快速入门:1.创建maven-web工程2.设......
  • springMvc报错
    这个报错困扰了我大概一天,主要是刚开始没抓到主要原因,是因为自己的项目结构出现了问题,导致找不到应有的东西,另一方面就是maven的问题,将maven解决后这个就能用了。具体......