Technologieaustausch

한어Русский языкEnglishFrançaisIndonesianSanskrit日本語DeutschPortuguêsΕλληνικάespañolItalianoSuomalainenLatina

RabbitMQ ist ein weit verbreiteter Open-Source-Nachrichtenbroker, der mehrere Messaging-Protokolle unterstützt und für zuverlässiges Messaging in verteilten Systemen verwendet werden kann. Zusätzlich zur grundlegenden Nachrichtenwarteschlangenfunktionalität bietet RabbitMQ auch erweiterte Funktionen, die seine Fähigkeiten in Bezug auf hohe Verfügbarkeit, Skalierbarkeit und Flexibilität verbessern. Hier sind einige der wichtigsten erweiterten Funktionen:

1. Hohe Verfügbarkeit

• Gespiegelte Warteschlangen：
  RabbitMQ bietet eine Spiegelwarteschlangenfunktion, um eine hohe Verfügbarkeit der Warteschlange zu erreichen, indem der Status und die Nachrichten der Warteschlange auf mehrere Knoten repliziert werden. Bei einem Ausfall des Master-Knotens ist ein nahtloser Wechsel zum Replikat-Knoten in der Spiegelwarteschlange möglich.

• Cluster-Modus：
  RabbitMQ kann im Cluster-Modus ausgeführt werden und Warteschlangen und Austauscher auf mehrere Knoten verteilen, um die Systemverfügbarkeit und Skalierbarkeit zu verbessern. Knoten im Cluster können miteinander kommunizieren und Nachrichten- und Warteschlangenmetadaten austauschen.

2. Nachrichtenkonsistenz

• Nachrichtenbestätigungen：
  Verbraucher können verarbeitete Nachrichten bestätigen, um sicherzustellen, dass sie nicht verloren gehen. Wenn die Nachricht nicht bestätigt wird, stellt RabbitMQ sie zur Verarbeitung durch andere Verbraucher wieder in die Warteschlange.

• Transaktionen：
  RabbitMQ unterstützt den AMQP-Transaktionsmodus, der es Produzenten ermöglicht, Nachrichten zu veröffentlichen und Nachrichten innerhalb einer Transaktion zu bestätigen, um die Atomizität und Konsistenz von Nachrichten sicherzustellen.

3. Haltbarkeit der Nachricht

• Permanente Nachrichten：
  Mit RabbitMQ können Nachrichten als dauerhaft markiert werden, um sicherzustellen, dass Nachrichten nach dem Neustart des Brokers nicht verloren gehen. Permanente Nachrichten werden auf die Festplatte geschrieben und nicht nur im Speicher gespeichert.

• Dauerhafte Warteschlangen：
  Die persistente Warteschlange bleibt auch nach dem Neustart des Brokers bestehen und stellt so sicher, dass die Metadaten der Warteschlange nicht verloren gehen.

4. Hoher Durchsatz und Parallelität

• Chargenbestätigungen：
  Ermöglichen Sie Verbrauchern, Nachrichten stapelweise zu bestätigen, reduzieren Sie den Netzwerk- und E/A-Overhead und verbessern Sie den Durchsatz.

• Anzahl der Vorabrufe：
  Durch Festlegen der Vorabrufanzahl kann der Verbraucher die gleichzeitige Verarbeitung von Nachrichten steuern, indem er neue Nachrichten aus der Warteschlange abruft, nachdem eine bestimmte Anzahl von Nachrichten verarbeitet wurde.

5. Plugins und Erweiterungen

• Plugin-System：
  RabbitMQ bietet ein flexibles Plug-In-System, und Benutzer können Plug-Ins laden und entladen, um die Funktionalität zu erhöhen. Beispielsweise wird das Shovel-Plugin verwendet, um Nachrichten über Cluster hinweg weiterzuleiten, und das Federation-Plugin wird verwendet, um Nachrichten zu übermitteln, die über geografische Standorte verteilt sind.

• Verwaltungs-Plugin：
  Bietet eine webbasierte Benutzeroberfläche zur Überwachung und Verwaltung von RabbitMQ-Instanzen, einschließlich der Anzeige des Warteschlangenstatus, der Switch-Konfiguration, der Nachrichtenrate usw.

6. Sicherheit

• TLS/SSL-Verschlüsselung：
  RabbitMQ unterstützt die Verwendung von TLS/SSL zur Verschlüsselung der Nachrichtenübertragung, um die Sicherheit der Nachrichten während der Übertragung zu gewährleisten.

• Zugangskontrolle：
  RabbitMQ bietet einen Zugriffskontrollmechanismus basierend auf Benutzern, Rollen und Berechtigungen, der es Administratoren ermöglicht, fein abgestufte Zugriffsberechtigungen zu konfigurieren.

7. Nachrichtenweiterleitung und -austausch

• Verschiedene Arten von Börsen (Börsen)：
  RabbitMQ unterstützt mehrere Arten von Switches, einschließlich Direct-, Topic-, Fanout- und Headers-Switches, um unterschiedliche Anforderungen an das Nachrichtenrouting zu erfüllen.

• Bindungen：
  Verbinden Sie Warteschlangen und Austausch über Bindungen, um komplexe Strategien zur Nachrichtenweiterleitung zu implementieren.

8. Überwachung und Management

• Überwachungsindikatoren (Metriken)：
  RabbitMQ bietet detaillierte Überwachungsindikatoren, einschließlich Nachrichtenrate, Warteschlangenlänge, Anzahl der Verbindungen usw., um Administratoren zu helfen, den Betriebsstatus des Systems zu verstehen.

• Alarme und Benachrichtigungen：
  RabbitMQ kann Alarme so konfigurieren, dass Benachrichtigungen ausgelöst werden, wenn die Warteschlangenlänge einen Schwellenwert überschreitet oder ein Knoten ausfällt.

9. Nachrichtenwiederholungs- und Warteschlangen für unzustellbare Nachrichten (Wiederholungs- und Warteschlangen für unzustellbare Nachrichten)

• Austausch unzustellbarer Nachrichten und Warteschlangen：
  Wenn eine Nachricht nicht verarbeitet werden kann oder die Anzahl der Wiederholungsversuche überschritten wird, kann sie zur weiteren Verarbeitung an die Warteschlange für unzustellbare Nachrichten weitergeleitet werden.

• Nachrichtenwiederholung：
  Unterstützt die Konfiguration von Nachrichtenwiederholungsstrategien, um sicherzustellen, dass die Nutzung wiederholt werden kann, wenn die Nutzung fehlschlägt.

10. Hybrid Cloud und Cross-Datacenter

• Rechenzentrumsübergreifende Replikation：
  Durch Plug-Ins oder manuelle Konfiguration unterstützt RabbitMQ die Nachrichtenreplikation zwischen verschiedenen Rechenzentren und gewährleistet so eine hohe Datenverfügbarkeit und Disaster-Recovery-Funktionen.

Diese erweiterten Funktionen machen RabbitMQ zu einer leistungsstarken und flexiblen Messaging-Middleware, die für verschiedene komplexe verteilte Systeme und Anwendungsszenarien geeignet ist. Durch die rationelle Nutzung dieser Funktionen kann ein leistungsstarkes, hochverfügbares und skalierbares Nachrichtensystem aufgebaut werden.

So implementieren Sie allgemeine erweiterte Funktionen in Spring:

1. Installation und Konfiguration

Stellen Sie zunächst sicher, dass Sie die Spring AMQP-Abhängigkeit in Ihr Projekt eingeführt haben:

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
1
2
3
4

existieren application.properties Konfigurieren Sie die RabbitMQ-Verbindungsinformationen in der Datei:

spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
1
2
3
4

2. Deklarieren Sie Warteschlangen, Austauscher und Bindungen

In Spring können Warteschlangen, Austauscher und Bindungsbeziehungen über @Bean definiert werden:

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
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
@EnableRabbit
public class RabbitConfig {

    static final String queueName = "testQueue";
    static final String exchangeName = "testExchange";

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

    @Bean
    DirectExchange exchange() {
        return new DirectExchange(exchangeName);
    }

    @Bean
    Binding binding(Queue queue, DirectExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("testRoutingKey");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

3. Bestätigung der Nachricht

Manuelle Nachrichtenbestätigung

Verbraucher können Nachrichten manuell bestätigen, um die Zuverlässigkeit der Nachrichtenverarbeitung sicherzustellen.verwenden@RabbitListener Beim Kommentieren können Sie konfigurierenacknowledgeMode fürMANUAL, und bestätigen Sie die Meldung manuell in der Methode:

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.support.Acknowledgment;

public class RabbitMQReceiver {

    @RabbitListener(queues = "testQueue", ackMode = "MANUAL")
    public void receiveMessage(Message message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
        try {
            // 处理消息
            System.out.println("Received message: " + new String(message.getBody()));
            
            // 手动确认消息
            channel.basicAck(tag, false);
        } catch (Exception e) {
            // 拒绝消息
            channel.basicNack(tag, false, true);
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

4. Nachrichtentransaktion

Implementieren Sie die Transaktionsunterstützung über RabbitTemplate:

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

@Service
public class RabbitMQService {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Transactional
    public void sendMessage(String message) {
        // 发送消息
        rabbitTemplate.convertAndSend("testExchange", "testRoutingKey", message);

        // 模拟事务回滚
        if (message.contains("error")) {
            throw new RuntimeException("Error occurred");
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

5. Warteschlange für unzustellbare Nachrichten

Konfigurieren Sie die Warteschlange für unzustellbare Nachrichten und ihre Bindungen:

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

@Bean
Binding dlqBinding() {
    return BindingBuilder.bind(dlq()).to(exchange()).with("dlqRoutingKey");
}

@Bean
Queue mainQueue() {
    Map<String, Object> args = new HashMap<>();
    args.put("x-dead-letter-exchange", exchangeName);
    args.put("x-dead-letter-routing-key", "dlqRoutingKey");
    return new Queue("mainQueue", true, false, false, args);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

6. Verzögerungswarteschlange

Verwenden Sie ein Plug-in, um eine Verzögerungswarteschlange zu implementieren, und Sie können eine verzögerte Nachrichtenzustellung erreichen, indem Sie die TTL (Time To Live) der Nachricht konfigurieren:

@Bean
Queue delayedQueue() {
    Map<String, Object> args = new HashMap<>();
    args.put("x-message-ttl", 60000); // 消息的 TTL 为 60 秒
    args.put("x-dead-letter-exchange", exchangeName);
    args.put("x-dead-letter-routing-key", "dlqRoutingKey");
    return new Queue("delayedQueue", true, false, false, args);
}
1
2
3
4
5
6
7
8

7. Gleichzeitige Verbraucher

über die Konfiguration SimpleRabbitListenerContainerFactory Implementieren Sie gleichzeitige Verbraucher:

import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleRabbitListenerContainerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitConfig {

    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setConcurrentConsumers(3); // 并发消费者数量
        factory.setMaxConcurrentConsumers(10);
        return factory;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

8. Plug-Ins und Erweiterungen

Nutzen Sie die Plug-in-Funktionen von RabbitMQ, z. B. die Verwendung des Shovel-Plug-ins zum Weiterleiten von Cluster-übergreifenden Nachrichten oder die Verwendung des Management-Plug-ins zur Überwachung und Verwaltung.