标签 mq 下的文章

Spring Cloud入门教程-使用rabbitMQ传输链路数据，储存到MySQL

2024-09-04

以下是一个简化的代码示例，展示如何使用Spring Cloud Stream与RabbitMQ接收数据，并将数据存储到MySQL数据库中。

添加依赖到pom.xml：

<dependencies>
    <!-- Spring Cloud Stream -->
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
    </dependency>
    <!-- MySQL Connector -->
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <scope>runtime</scope>
    </dependency>
    <!-- Spring Data JPA -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-jpa</artifactId>
    </dependency>
</dependencies>

配置application.yml：

spring:
  cloud:
    stream:
      binders:
        rabbit:
          type: rabbit
          environment:
            spring:
              rabbitmq:
                host: localhost
                port: 5672
                username: guest
                password: guest
      bindings:
        input:
          destination: data-input
          content-type: application/json
          binder: rabbit
  datasource:
    url: jdbc:mysql://localhost:3306/testdb
    username: dbuser
    password: dbpass
    driver-class-name: com.mysql.cj.jdbc.Driver
  jpa:
    show-sql: true
    database-platform: org.hibernate.dialect.MySQL5InnoDBDialect

创建实体类LinkData.java：

@Entity
public class LinkData {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    private String url;
    // getters and setters
}

创建仓库接口LinkDataRepository.java：

public interface LinkDataRepository extends JpaRepository<LinkData, Long> {}

创建消息接收器LinkDataProcessor.java：

@EnableBinding(Sink.class)
public class LinkDataProcessor {
    @Autowired
    private LinkDataRepository repository;

    @StreamListener(Sink.INPUT)
    public void handle(LinkData linkData) {
        repository.save(linkData);
    }
}

启动类Application.java：

@SpringBootApplication
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(App

System

2024-09-04

所有,java

@Configuration
public class RocketMQConfig {

    @Bean
    public RocketMQListener rocketMQListener() {
        // 创建并配置RocketMQ监听器
        RocketMQListener listener = new RocketMQListener();
        listener.setConsumerGroup("your_consumer_group");
        listener.setNameServer("your_name_server");
        listener.setTopic("your_topic");
        // 可以设置更多的监听器属性
        return listener;
    }
}

这个配置类创建了一个RocketMQListener的Bean，并设置了必要的属性，包括消费者组、名称服务器地址和主题。在Spring应用上下文中启用后，这个监听器将会自动注册到RocketMQ，并开始接收消息。

- 阅读更多 -

【SpringBoot高级篇】SpringBoot集成RocketMQ消息队列

System

2024-09-04

所有,java

为了在Spring Boot应用程序中集成RocketMQ，你需要按照以下步骤操作：

添加依赖：在pom.xml中添加RocketMQ的Spring Boot Starter依赖。

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.0.3</version>
</dependency>

配置RocketMQ：在application.properties或application.yml中配置RocketMQ的基本属性。

# application.properties
spring.rocketmq.name-server=127.0.0.1:9876
spring.rocketmq.producer.group=my-group

发送消息：创建一个Service来发送消息。

import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class RocketMQService {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    public void sendMessage(String topic, String tag, String message) {
        rocketMQTemplate.send(topic, tag, message);
    }
}

接收消息：创建一个Listener来接收消息。

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

@Component
@RocketMQMessageListener(topic = "your-topic", consumerGroup = "your-consumer_group")
public class ConsumerListener implements RocketMQListener<String> {

    @Override
    public void onMessage(String message) {
        // 处理接收到的消息
        System.out.println("Received message: " + message);
    }
}

确保你的RocketMQ服务器正在运行，并且你的应用程序配置正确指向它。上述代码提供了一个简单的例子，展示了如何在Spring Boot应用程序中发送和接收RocketMQ消息。

- 阅读更多 -

SpringCloud-MQ消息队列

System

2024-09-04

所有,java

Spring Cloud Stream 是一个构建消息驱动微服务的框架。它整合了消息中间件，使得在 Spring Boot 应用中发送和接收消息变得容易。

以下是一个使用 Spring Cloud Stream 从源发送消息的简单例子：

添加依赖到你的 pom.xml：

<dependencies>
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
    </dependency>
</dependencies>

<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-dependencies</artifactId>
            <version>Finchley.SR2</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>

配置 application.yml 或 application.properties：

spring:
  cloud:
    stream:
      binders:
        defaultRabbit:
          type: rabbit
          environment:
            spring:
              rabbitmq:
                host: localhost
                port: 5672
                username: guest
                password: guest
      bindings:
        output:
          destination: test-topic
          content-type: application/json
          binder: defaultRabbit

发送消息的服务类：

@EnableBinding(Source.class)
public class MessageSender {

    @Autowired
    private MessageChannel output;

    public void send(String message) {
        this.output.send(MessageBuilder.withPayload(message).build());
    }
}

使用发送器发送消息：

@RestController
public class TestController {

    @Autowired
    private MessageSender messageSender;

    @GetMapping("/sendMessage")
    public String sendMessage(@RequestParam String message) {
        messageSender.send(message);
        return "Message sent";
    }
}

在这个例子中，我们定义了一个名为 output 的消息通道，并将其绑定到名为 test-topic 的消息主题。我们还配置了 Spring Cloud Stream 使用默认的 RabbitMQ 绑定器。MessageSender 类中的 send 方法用于将消息发送到绑定的通道。TestController 中的 sendMessage 方法通过 HTTP GET 请求触发消息发送。

- 阅读更多 -

SpringBoot中整合RabbitMQ（测试+部署上线最完整）

System

2024-09-04

所有,java

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.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitMQConfig {

    private static final String TOPIC_EXCHANGE = "topicExchange";
    private static final String QUEUE_NAME = "queueName";

    @Bean
    Queue queue() {
        return new Queue(QUEUE_NAME, true);
    }

    @Bean
    TopicExchange exchange() {
        return new TopicExchange(TOPIC_EXCHANGE);
    }

    @Bean
    Binding binding(Queue queue, TopicExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("routingKey");
    }

    @Bean
    SimpleMessageListenerContainer container(ConnectionFactory connectionFactory,
                                            MessageListenerAdapter listenerAdapter) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setQueueNames(QUEUE_NAME);
        container.setMessageListener(listenerAdapter);
        return container;
    }

    @Bean
    MessageListenerAdapter listenerAdapter(RabbitMQListener receiver) {
        return new MessageListenerAdapter(receiver, "handleMessage");
    }
}

public class RabbitMQListener {
    public void handleMessage(String message) {
        // 处理接收到的消息
    }
}

这个代码示例展示了如何在Spring Boot应用程序中配置和连接RabbitMQ，创建交换器、队列并将它们绑定在一起。同时，它还配置了一个监听器容器来监听指定队列的消息，并将接收到的消息委托给一个处理器（RabbitMQListener）进行处理。这个例子涵盖了RabbitMQ在Spring Boot中的基本使用，并且是在生产环境中部署RabbitMQ的一个很好的实践。

- 阅读更多 -

Spring Boot集成RabbitMQ

System

2024-09-04

所有,java

要在Spring Boot中集成RabbitMQ，你需要做以下几步：

添加依赖：在pom.xml中添加Spring Boot的RabbitMQ依赖。

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

配置RabbitMQ：在application.properties或application.yml中配置RabbitMQ连接信息。

# application.properties
spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

创建配置类：配置队列、交换器、路由等。

@Configuration
public class RabbitMQConfig {

    @Bean
    Queue myQueue() {
        return new Queue("myQueue", true);
    }

    @Bean
    DirectExchange myExchange() {
        return new DirectExchange("myExchange");
    }

    @Bean
    Binding binding(Queue myQueue, DirectExchange myExchange) {
        return BindingBuilder.bind(myQueue).to(myExchange).with("myRoutingKey");
    }
}

发送和接收消息：使用RabbitTemplate发送消息，使用@RabbitListener注解接收消息。

@Service
public class RabbitMQService {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendMessage(String message) {
        rabbitTemplate.convertAndSend("myExchange", "myRoutingKey", message);
    }
}

@Component
public class RabbitMQListener {

    @RabbitListener(queues = "myQueue")
    public void receiveMessage(String message) {
        System.out.println("Received message: " + message);
    }
}

确保你的RabbitMQ服务器正在运行，并且你的Spring Boot应用程序可以连接到它。以上代码提供了一个简单的例子，展示了如何在Spring Boot应用程序中集成RabbitMQ。

- 阅读更多 -

SpringCloud-基于SpringAMQP实现消息队列

System

2024-09-03

所有,java

在Spring Cloud中，基于Spring AMQP的消息队列实现通常涉及以下步骤：

配置RabbitMQ服务器信息。
创建交换器（Exchange）和队列（Queue）。
使用@RabbitListener注解创建消息消费者。
使用RabbitTemplate发送消息。

以下是一个简单的例子：

配置文件application.yml:

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

配置类RabbitConfig.java:

@Configuration
public class RabbitConfig {

    @Bean
    Queue myQueue() {
        return new Queue("myQueue", true);
    }

    @Bean
    DirectExchange myExchange() {
        return new DirectExchange("myExchange");
    }

    @Bean
    Binding binding(Queue myQueue, DirectExchange myExchange) {
        return BindingBuilder.bind(myQueue).to(myExchange).with("myRoutingKey");
    }
}

消息消费者MessageConsumer.java:

@Component
public class MessageConsumer {

    @RabbitListener(queues = "myQueue")
    public void handleMessage(String message) {
        System.out.println("Received message: " + message);
    }
}

消息生产者MessageProducer.java:

@Component
public class MessageProducer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendMessage(String message) {
        rabbitTemplate.convertAndSend("myExchange", "myRoutingKey", message);
    }
}

在这个例子中，我们定义了一个名为myQueue的队列，一个名为myExchange的直接交换器，并将它们通过myRoutingKey绑定起来。MessageConsumer类中的handleMessage方法会接收队列中的消息。MessageProducer类中的sendMessage方法会使用RabbitTemplate发送消息到交换器。

确保你的环境中已经安装并运行了RabbitMQ服务，并且在发送和接收消息之前，队列和交换器应该已经被正确配置。

- 阅读更多 -

SpringCloud学习笔记RabbitMQ

System

2024-09-03

所有,java

RabbitMQ是一个开源的消息代理和队列服务器，用来通过插件机制来支持多种消息协议，并且可以支持多种语言的客户端，如Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等。

在Spring Cloud中，我们可以使用Spring-AMQP和RabbitMQ来实现消息队列服务。

使用RabbitTemplate发送消息

@Autowired
private RabbitTemplate rabbitTemplate;

public void send() {
    String context = "hello " + new Date();
    System.out.println("Sender : " + context);

    this.rabbitTemplate.convertAndSend("hello", context);
}

使用@RabbitListener注解创建消息消费者

@RabbitListener(queues = "hello")
public void receive(String content) {
    System.out.println("Receiver : " + content);
}

使用@EnableRabbit注解开启Spring AMQP功能

@Configuration
@EnableRabbit
public class RabbitConfig {

    @Bean
    Queue queue() {
        return new Queue("hello", true);
    }
}

使用RabbitMQ实现异步下单

@Autowired
private RabbitTemplate rabbitTemplate;

public void order(){
    String context = "hello " + new Date();
    System.out.println("Sender : " + context);

    this.rabbitTemplate.convertAndSend("order", context);
}

@RabbitListener(queues = "order")
public void receiveOrder(String content) {
    System.out.println("Receiver order : " + content);
}

以上代码就是使用Spring Cloud结合RabbitMQ的一个简单示例。

注意：在实际开发中，我们还需要对RabbitMQ的连接、消费者、队列等进行配置，并且要考虑到消息的确认机制、重试机制、持久化等问题，以确保系统的稳定性和可靠性。

- 阅读更多 -

GoLang学习-Redis实现MQ功能

System

2024-09-03

所有,redis

package main

import (
    "fmt"
    "github.com/go-redis/redis"
    "log"
    "os"
    "os/signal"
    "syscall"
    "time"
)

func main() {
    client := redis.NewClient(&redis.Options{
        Addr:     "localhost:6379",
        Password: "", // no password set
        DB:       0,  // use default DB
    })

    pong, err := client.Ping().Result()
    if err != nil {
        log.Fatalf("Failed to connect to Redis: %v", err)
    }
    fmt.Println(pong)

    // 创建一个消息队列的频道
    messages := make(chan string)

    // 创建一个信号通知，用于优雅地关闭程序
    sigs := make(chan os.Signal, 1)
    signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)

    // 开始监听消息队列
    go func() {
        for {
            // 从Redis的列表中取出消息
            val, err := client.BLPop(0, "myqueue").Result()
            if err != nil {
                log.Println("Error receiving message:", err)
                continue
            }
            messages <- val[1]
        }
    }()

    // 循环处理消息
    for {
        select {
        case msg := <-messages:
            // 处理消息
            log.Printf("Received: %s\n", msg)
        case sig := <-sigs:
            log.Printf("Received signal: %s. Exiting...\n", sig)
            return
        }
    }
}

这段代码使用Go语言和go-redis库实现了一个简单的Redis消息队列客户端。它连接到Redis服务器，监听名为"myqueue"的列表，并将接收到的消息打印出来。如果接收到系统信号（如SIGINT或SIGTERM），程序会优雅地关闭。这个例子展示了如何使用Go语言处理Redis消息队列，并且是构建实时系统时理解消息队列模式的一个很好的起点。

- 阅读更多 -

SpringBoot程序MQ宕机备用方案-多活架构

System

2024-09-03

所有,java

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Configuration
public class MqConfig {

    @Bean(name = "primaryMqExecutor")
    @Primary
    public ExecutorService primaryMqExecutor() {
        return new ThreadPoolExecutor(
                5,
                20,
                300,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(1024),
                new CustomizableThreadFactory("primary-mq-thread-"),
                new ThreadPoolExecutor.CallerRunsPolicy()
        );
    }

    @Bean(name = "standbyMqExecutor")
    public ExecutorService standbyMqExecutor() {
        return new ThreadPoolExecutor(
                5,
                20,
                300,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(1024),
                new CustomizableThreadFactory("standby-mq-thread-"),
                new ThreadPoolExecutor.CallerRunsPolicy()
        );
    }
}

这个代码示例定义了两个ExecutorService Bean，一个用于主消息队列（primaryMqExecutor），另一个用于备用消息队列（standbyMqExecutor）。通过定义两个Bean，我们可以在程序中根据主消息队列的状态动态地切换到备用消息队列，实现消息队列的多活（Multi-Active）架构。这里使用了ThreadPoolExecutor来创建线程池，并通过CustomizableThreadFactory来设置线程名的前缀，使得线程的功能更加清晰。

- 阅读更多 -