Analyse häufiger Probleme und Lösungen der Java RabbitMQ-Nachrichtenwarteschlange

Nachrichtenakkumulation

Das Generierungsszenario der Nachrichtenakkumulation:

• Die Geschwindigkeit der von Produzenten generierten Nachrichten ist größer als die Geschwindigkeit von Verbraucherkonsum. Lösung: Erhöhen Sie die Anzahl oder Geschwindigkeit der Verbraucher.

• Wenn es keine Verbraucher zum Konsumieren gibt. Lösung: Warteschlange für unzustellbare Nachrichten, Gültigkeitsdauer der Nachricht festlegen. Dies entspricht dem Festlegen eines Gültigkeitszeitraums für unsere Nachrichten. Wenn innerhalb der angegebenen Zeit kein Verbrauch erfolgt, wird die Client-Rückrufüberwachungsmethode ausgeführt, um die Nachricht im Datenbanktabellendatensatz zu speichern Die Entschädigung wird später realisiert.

Stellen Sie sicher, dass die Nachricht nicht verloren geht

1 Der Produzent verwendet den Nachrichtenbestätigungsmechanismus, um sicherzustellen, dass die Nachricht an MQ übermittelt werden kann erfolgreich.

2. Der MQ-Server sollte die Nachricht auf der Festplatte speichern

3 Der Verbraucher verwendet den manuellen Bestätigungsmechanismus, um zu bestätigen, dass der Nachrichtenverbrauch erfolgreich ist

Wenn MQ Was soll ich tun, wenn die Serverkapazität voll ist?

Verwenden Sie die Warteschlange für unzustellbare Nachrichten, um Nachrichten in der Datenbank zu speichern und den späteren Verbrauch auszugleichen.

Dead-Letter-Warteschlange

RabbitMQ-Dead-Letter-Warteschlange wird allgemein als Ersatzreifenwarteschlange bezeichnet. Nachdem die Nachrichten-Middleware die Nachricht aus irgendeinem Grund abgelehnt hat, kann sie an die Dead-Letter-Warteschlange weitergeleitet werden Briefwarteschlange zur Speicherung, die Warteschlange für unzustellbare Briefe kann auch Schalter- und Routing-Tasten usw. haben.

Generierungshintergrund:

• Die Nachricht wird an MQ übermittelt und gespeichert. Die Nachricht ist abgelaufen.

#🎜 🎜##🎜 🎜#Die Warteschlange erreicht die maximale Länge (der Warteschlangencontainer ist voll) und der Produzent weigert sich, die Nachricht zu empfangen Nachrichten werden in die Warteschlange für unzustellbare Nachrichten verschoben Konfiguration

• <dependencies>
          <!-- springboot-web组件 -->
          <dependency>
              <groupId>org.springframework.boot</groupId>
              <artifactId>spring-boot-starter-web</artifactId>
          </dependency>
          <!-- 添加springboot对amqp的支持 -->
          <dependency>
              <groupId>org.springframework.boot</groupId>
              <artifactId>spring-boot-starter-amqp</artifactId>
          </dependency>
          <dependency>
              <groupId>org.apache.commons</groupId>
              <artifactId>commons-lang3</artifactId>
          </dependency>
          <!--fastjson -->
          <dependency>
              <groupId>com.alibaba</groupId>
              <artifactId>fastjson</artifactId>
              <version>1.2.49</version>
          </dependency>
      </dependencies>

  #🎜🎜 #Queue-Konfigurationsklasse

  server:
  #  服务启动端口配置
    port: 8081
    servlet:
  #    应用访问路径
      context-path: /
  spring:
    #增加application.druid.yml 的配置文件
  #  profiles:
  #    active: rabbitmq
    rabbitmq:
      ####连接地址
      host: www.kaicostudy.com
      ####端口号
      port: 5672
      ####账号
      username: kaico
      ####密码
      password: kaico
      ### 地址
      virtual-host: /kaicoStudy
  ###模拟演示死信队列
  kaico:
    dlx:
      exchange: kaico_order_dlx_exchange
      queue: kaico_order_dlx_queue
      routingKey: kaico.order.dlx
    ###备胎交换机
    order:
      exchange: kaico_order_exchange
      queue: kaico_order_queue
      routingKey: kaico.order

  Dead-Letter-Warteschlange-Verbraucher

@Configuration
public class DeadLetterMQConfig {
    /**
     * 订单交换机
     */
    @Value("${kaico.order.exchange}")
    private String orderExchange;
    /**
     * 订单队列
     */
    @Value("${kaico.order.queue}")
    private String orderQueue;
    /**
     * 订单路由key
     */
    @Value("${kaico.order.routingKey}")
    private String orderRoutingKey;
    /**
     * 死信交换机
     */
    @Value("${kaico.dlx.exchange}")
    private String dlxExchange;
    /**
     * 死信队列
     */
    @Value("${kaico.dlx.queue}")
    private String dlxQueue;
    /**
     * 死信路由
     */
    @Value("${kaico.dlx.routingKey}")
    private String dlxRoutingKey;
    /**
     * 声明死信交换机
     *
     * @return DirectExchange
     */
    @Bean
    public DirectExchange dlxExchange() {
        return new DirectExchange(dlxExchange);
    }
    /**
     * 声明死信队列
     *
     * @return Queue
     */
    @Bean
    public Queue dlxQueue() {
        return new Queue(dlxQueue);
    }
    /**
     * 声明订单业务交换机
     *
     * @return DirectExchange
     */
    @Bean
    public DirectExchange orderExchange() {
        return new DirectExchange(orderExchange);
    }
    /**
     * 绑定死信队列到死信交换机
     *
     * @return Binding
     */
    @Bean
    public Binding binding() {
        return BindingBuilder.bind(dlxQueue())
                .to(dlxExchange())
                .with(dlxRoutingKey);
    }
    /**
     * 声明订单队列，并且绑定死信队列
     *
     * @return Queue
     */
    @Bean
    public Queue orderQueue() {
        // 订单队列绑定我们的死信交换机
        Map<String, Object> arguments = new HashMap<>(2);
        arguments.put("x-dead-letter-exchange", dlxExchange);
        arguments.put("x-dead-letter-routing-key", dlxRoutingKey);
        return new Queue(orderQueue, true, false, false, arguments);
    }
    /**
     * 绑定订单队列到订单交换机
     *
     * @return Binding
     */
    @Bean
    public Binding orderBinding() {
        return BindingBuilder.bind(orderQueue())
                .to(orderExchange())
                .with(orderRoutingKey);
    }
}

• Gewöhnlicher Warteschlangen-Verbraucher

  @Component
  public class OrderDlxConsumer {
      /**
       * 死信队列监听队列回调的方法
       * @param msg
       */
      @RabbitListener(queues = "kaico_order_dlx_queue")
      public void orderDlxConsumer(String msg) {
          System.out.println("死信队列消费订单消息" + msg);
      }
  }

  Die Hintergrundwarteschlange Die Verwaltungsseite lautet wie folgt:

Bereitstellungsmethode: Die Warteschlange für unzustellbare Nachrichten darf nicht auf demselben Server wie die normale Warteschlange vorhanden sein und sollte auf separaten Servern gespeichert werden .

Delay Queue

Implementierungsplan für das automatische Timeout und Schließen des Systems, wenn die Bestellung 30 Minuten lang nicht bezahlt wird.

Basierend auf der Aufgabenplanung ist die Effizienz sehr gering.

Basierend auf der Redis-Implementierung eines abgelaufenen Schlüssels wird eine Methode an den Client zurückgerufen, wenn der Schlüssel abläuft.

Wenn ein Benutzer eine Bestellung aufgibt, wird ein Token (Gültigkeitszeitraum) für 30 Minuten generiert und in unserem Redis gespeichert. Nachteile: Sehr redundant, es wird ein redundantes Feld in der Tabelle gespeichert.

MQ-basierte Verzögerungswarteschlange (beste Lösung) Rabbitmq-Fall.

Prinzip: Wenn wir eine Bestellung aufgeben, übermitteln wir eine Nachricht an mq und legen die Gültigkeitsdauer auf 30 Minuten fest. Wenn die Nachricht jedoch abläuft (ohne verbraucht zu werden), führen wir eine Methode auf unserem Client aus, um dies mitzuteilen Unsere Nachricht ist abgelaufen. Zu diesem Zeitpunkt prüfen wir, ob die Bestellung bezahlt wurde.

Implementierungslogik:

Zur Implementierung wird hauptsächlich die Warteschlange für unzustellbare Nachrichten verwendet.

Der gewünschte Code: Normale Verbraucher konsumieren keine Nachrichten, oder es gibt keine normalen Verbraucher und treten nach der festgelegten Zeit in die Warteschlange für unzustellbare Nachrichten ein Der Dead-Letter-Consumer implementiert die entsprechende Geschäftslogik. { Der Standardwert ist unzählige Wiederholungsversuche)

Die Anzahl der RabbitMQ-Wiederholungsversuche sollte begrenzt sein, z. B. maximal 5 Wiederholungsversuche, mit einem Intervall von jeweils 3 Sekunden; wenn der Wiederholungsversuch mehrmals fehlschlägt, wird dies der Fall sein Bis zum Tod gespeichert oder in der Datenbanktabelle gespeichert, um eine spätere manuelle Kompensation aufzuzeichnen. Denn nach der Anzahl der fehlgeschlagenen Wiederholungsversuche löscht die Warteschlange die Nachricht automatisch.

Prinzip der Nachrichtenwiederholung: Verwenden Sie während des Wiederholungsprozesses aop, um unsere Verbrauchsüberwachungsmethode abzufangen, und dieses Fehlerprotokoll wird nicht gedruckt. Wenn der Vorgang nach mehreren Wiederholungsversuchen fehlschlägt, wird das Fehlerprotokoll erst gedruckt, wenn die maximale Anzahl von Fehlern erreicht ist.

Wenn der Konsum nach mehrmaligem Konsum fehlschlägt:

1. Nachricht automatisch löschen; (die Nachricht geht möglicherweise verloren)

Lösung: # 🎜 🎜#

Wenn der Nachschub mehrmals fehlschlägt, wird er schließlich in der Warteschlange für unzustellbare Nachrichten gespeichert.

verwendet die Tabellenprotokollierung, um Verbrauchsfehler-Fehlerprotokolle aufzuzeichnen, die später manuell verarbeitet werden Umsetzungsentschädigung.

Angemessene Wahl des Wiederholungsmechanismus

Nachdem der Verbraucher die Nachricht erhalten hat, ruft er eine Schnittstelle eines Drittanbieters auf (HTTP-Anfrage), aber was ist, wenn der Anruf an den Drittanbieter erfolgt? Schnittstelle fällt aus? Müssen Sie es noch einmal versuchen?

Antwort: Manchmal schlägt der Anruf aufgrund einer Netzwerkausnahme fehl und muss möglicherweise mehrmals wiederholt werden.

Nachdem der Verbraucher die Nachricht erhalten hat, wird aufgrund eines Codeproblems eine Datenausnahme ausgelöst. Muss es erneut versucht werden?

Antwort: Es besteht keine Notwendigkeit, es erneut zu versuchen. Wenn der Code abnormal ist, muss das Code-Release-Projekt erneut geändert werden.

Verbraucher aktivieren den manuellen Bestätigungsmodus

Im ersten Schritt muss bei der Springboot-Projektkonfiguration der Bestätigungsmodus aktiviert werden

Bestätigungsmodus : Handbuch

Zweiter Schritt, Verbraucher-Java-Code

@Component
public class OrderConsumer {
    /**
     * 监听队列回调的方法
     *
     * @param msg
     */
    @RabbitListener(queues = "kaico_order_queue")
    public void orderConsumer(String msg) {
        System.out.println("正常订单消费者消息msg:" + msg);
    }
}

Wie RabbitMQ das Problem der Nachrichten-Idempotenz löst

Was ist Nachrichten-Idempotenz? Wie stellt der MQ-Verbraucher Idempotenz sicher?

Der Grund: Der Verbraucher aktiviert möglicherweise die automatische Wiederholung, und der Wiederholungsprozess kann dazu führen, dass der Geschäftslogikcode des Verbrauchers wiederholt ausgeführt wird. In diesem Moment wurde die Nachricht verbraucht. Da der Geschäftsfehler dazu geführt hat, dass die Nachricht erneut verbraucht wurde, wird

angezeigt. Verwenden Sie die globale ID der Nachricht, um dies anhand des Geschäfts zu bestimmen basierend auf der Geschäfts-ID (globale eindeutige ID). Diese Nachricht wurde verbraucht.

Consumer-Code-Logik:

RabbitMQ löst verteilte Transaktionsprobleme

Verteilte Transaktionen: In a In einem verteilten System gibt es aufgrund der dienstübergreifenden Aufrufschnittstelle mehrere verschiedene Transaktionen, und jede Transaktion beeinflusst sich nicht gegenseitig. Es gibt ein Problem verteilter Transaktionen.

Lösen Sie die Kernidee verteilter Transaktionen: endgültige Datenkonsistenz.

Substantiv im verteilten Feld:

Starke Konsistenz: Entweder ist die Synchronisationsgeschwindigkeit sehr hoch oder der Sperrmechanismus lässt keine schmutzigen Lesevorgänge zu; Starke Konsistenzlösung: Entweder synchronisiert Datenbank A die Daten sehr schnell mit Daten B, oder Datenbank B kann die Daten nicht lesen, bevor die Synchronisierung von Datenbank A abgeschlossen ist.

Schwache Konsistenz: Die Daten, die gelesen werden dürfen, sind die ursprünglichen schmutzigen Daten, und die gelesenen Ergebnisse dürfen inkonsistent sein.

Endgültige Konsistenz: Da in unserem verteilten System Daten synchron über das Netzwerk kommuniziert werden, sind kurze Datenverzögerungen zulässig, aber die endgültigen Daten müssen konsistent sein.

Die Idee, verteilte Transaktionen basierend auf RabbitMQ zu lösen

Die Idee, verteilte Transaktionen basierend auf RabbitMQ zu lösen: (Übernahme der endgültigen Konsistenzlösung)

# 🎜🎜#

Bestätigen Sie, dass unsere Produzentennachrichten an MQ zugestellt werden müssen (Nachrichtenbestätigungsmechanismus). Wenn die Zustellung fehlschlägt, versuchen Sie es erneut.

• #🎜🎜 #Verbraucher verwenden manuell Die Bestätigungsnachricht in Form einer Bestätigung erkennt den Verbrauch und berücksichtigt das Idempotenzproblem. Im Falle eines Verbrauchsfehlers hilft mq dem Verbraucher automatisch, es erneut zu versuchen.

• Stellen Sie sicher, dass die erste Transaktion unseres Produzenten zuerst ausgeführt wird. Wenn die Ausführung fehlschlägt, verwenden Sie die zusätzliche Auftragswarteschlange (ergänzen Sie die eigene Transaktion des Produzenten, um sicherzustellen, dass die erste Transaktion des Produzenten ausgeführt wird abgeschlossen) [Eventualkonsistenz der Daten]).

• Lösungskarte: Der Kern besteht darin, mit mq Nachrichten an andere Systeme zu senden, um die Daten wieder zu ändern.

Das obige ist der detaillierte Inhalt vonAnalyse häufiger Probleme und Lösungen der Java RabbitMQ-Nachrichtenwarteschlange. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!