RabbitMQ Messaging & Async Systems · บทเรียน

การตอบรับจากคอนซูเมอร์และการจัดเข้าคิวใหม่

บทเรียน 3 จาก 411 ขั้นตอน

การตอบรับจากคอนซูเมอร์และการจัดเข้าคิวใหม่ เป็นบทเรียน RabbitMQ Messaging & Async Systems ฟรีบน CoddyKit นี่คือบทเรียนที่ 3 จากทั้งหมด 4 บทเรียน คุณสามารถอ่านบทเรียนทั้งหมดด้านล่างฟรี — จากนั้นลองปฏิบัติด้วยตัวคุณเองในเบราว์เซอร์พร้อมตัวแก้ไขโค้ดในตัวและติวเตอร์ AI ตลอด 24/7 บทเรียนนี้เป็นส่วนหนึ่งของเส้นทางการเรียน RabbitMQ Messaging & Async Systems และความก้าวหน้าของคุณจะซิงค์ข้ามเว็บและแอป CoddyKit คอร์ส RabbitMQ Messaging & Async Systems มีบทเรียนทั้งหมด 4 บทเรียน

บางส่วนของบทเรียนนี้ยังไม่ได้รับการแปล และแสดงเป็นภาษาอังกฤษ

Reliable Message Consumption

In distributed systems, ensuring messages are processed correctly is vital. What happens if a consumer crashes mid-processing? Or if a message causes an error?

This lesson explores consumer acknowledgements and requeuing, essential techniques for building fault-tolerant message consumers.

ACKs: The Handshake

A consumer acknowledgement (ACK) is a signal sent by the consumer back to RabbitMQ. It tells the broker: "I've successfully received and processed this message."

Without an ACK, RabbitMQ assumes the message hasn't been processed.
This mechanism prevents message loss if a consumer fails before finishing its work.

Automatic vs. Manual ACKs

RabbitMQ offers two ways to acknowledge messages:

Automatic (autoAck=true): Messages are acknowledged immediately upon delivery to the consumer. Simple, but risky if processing fails.
Manual (autoAck=false): The consumer explicitly sends an ACK after successful processing. This is the default and recommended for reliable systems.

We'll focus on manual acknowledgements for robustness.

Sending a Message (Producer)

Let's set up a simple Java producer to send a message. This message will be consumed later, and we'll apply manual acknowledgements.

Run this code to send a "Hello RabbitMQ!" message.

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.nio.charset.StandardCharsets;

public class MyProducer {
    private final static String QUEUE_NAME = "ack_queue";

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost"); // Connect to local RabbitMQ

        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {

            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            String message = "Hello RabbitMQ!";
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes(StandardCharsets.UTF_8));
            System.out.println(" [x] Sent '" + message + "'");
        }
    }
}

Manual Acknowledgements in Action

Now, let's create a consumer that uses manual acknowledgements. Notice the autoAck parameter is set to false when consuming.

The channel.basicAck() call confirms successful processing.

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import java.nio.charset.StandardCharsets;

public class MyConsumerAck {
    private final static String QUEUE_NAME = "ack_queue";

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
            System.out.println(" [x] Received '" + message + "'");
            try {
                // Simulate processing work
                Thread.sleep(1000);
                System.out.println(" [x] Done processing '" + message + "'");
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); // Manual ACK!
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                System.err.println(" [!] Processing interrupted.");
                // In a real app, you might re-queue or handle error
            }
        };
        // autoAck is set to false for manual acknowledgements
        channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {});
    }
}

When Processing Fails

What if our consumer encounters an error during message processing? If we only use basicAck, the message is lost even if processing wasn't completed.

RabbitMQ provides ways to inform the broker that a message could not be processed successfully, giving us options to handle it.

Requeuing Failed Messages

If a message fails processing due to a transient error (e.g., database connection down), you might want to retry it later. Use channel.basicNack(deliveryTag, multiple, requeue) or channel.basicReject(deliveryTag, requeue).

Setting requeue to true sends the message back to the queue for another consumer to pick up.

Run this consumer. It will intentionally fail twice and then successfully process the message.

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;

public class MyConsumerRequeue {
    private final static String QUEUE_NAME = "ack_queue";
    private static int attemptCount = 0;

    public static void main(String[] argv) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
            long deliveryTag = delivery.getEnvelope().getDeliveryTag();
            System.out.println(" [x] Received '" + message + "' (Attempt: " + (++attemptCount) + ")");

            try {
                if (attemptCount <= 2) { // Simulate failure for first 2 attempts
                    throw new RuntimeException("Simulated processing error!");
                }
                // Simulate successful processing
                Thread.sleep(1000);
                System.out.println(" [x] Successfully processed '" + message + "'");
                channel.basicAck(deliveryTag, false); // ACK on success
            } catch (Exception e) {
                System.err.println(" [!] Error processing '" + message + "': " + e.getMessage());
                channel.basicNack(deliveryTag, false, true); // NACK and requeue!
                System.out.println(" [!] Message '" + message + "' requeued.");
            }
        };
        channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {});
    }
}

Discarding Problematic Messages

Sometimes, a message is fundamentally flawed and will *always* cause an error. Requeuing it repeatedly is pointless and can lead to a "poison message" loop.

In such cases, set requeue to false. This discards the message from the queue. Often, these messages are sent to a Dead Letter Exchange (DLX) for later inspection, but we'll cover DLX in a future lesson.

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;

public class MyConsumerReject {
    private final static String QUEUE_NAME = "ack_queue";

    public static void main(String[] argv) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
            long deliveryTag = delivery.getEnvelope().getDeliveryTag();
            System.out.println(" [x] Received '" + message + "'");

            // Simulate an unrecoverable error
            System.err.println(" [!] Fatal error for '" + message + "'. Discarding.");
            channel.basicNack(deliveryTag, false, false); // NACK and DO NOT requeue!
            System.out.println(" [!] Message '" + message + "' rejected (not requeued).");
        };
        channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {});
    }
}

Idempotent Consumers

When you requeue messages, there's a chance a consumer might process the same message multiple times.

It's crucial to design your consumers to be idempotent. This means processing the same message twice (or more) produces the same result as processing it once, without unintended side effects.

Reliability Check

Consider a RabbitMQ consumer configured with manual acknowledgements. If a message is received but the consumer crashes *before* calling channel.basicAck() or channel.basicNack(), what typically happens to that message?

ตอบคำถามในบทเรียนแบบโต้ตอบ

Recap: Building Reliable Consumers

You've learned how to make your RabbitMQ consumers resilient:

Manual Acknowledgements: Give consumers control over message fate.
Requeuing: Retry messages for transient failures using basicNack(..., true).
Discarding: Prevent poison message loops with basicNack(..., false).
Idempotency: Design consumers to handle duplicate deliveries gracefully.

These techniques are fundamental for building robust, fault-tolerant messaging systems.

เริ่มต้นได้ฟรี

เรียนรู้ RabbitMQ Messaging & Async Systems ด้วย AI tutor — ฟรี

เขียนและเรียกใช้โค้ดจริงในเบราว์เซอร์ของคุณ รับความช่วยเหลือทันทีจาก AI tutor 24/7 และเรียนรู้ต่อจากที่คุณหยุดบนเว็บหรือในแอป

คอร์ส: 11
บทเรียน: 44

เริ่มเรียนฟรี ดูหลักสูตรเต็ม

คำถามที่พบบ่อย

บทเรียน “การตอบรับจากคอนซูเมอร์และการจัดเข้าคิวใหม่” ฟรีหรือไม่

ใช่ — ข้อความเต็มของ “การตอบรับจากคอนซูเมอร์และการจัดเข้าคิวใหม่” ฟรีให้อ่านที่นี่บนเว็บ เพื่อปฏิบัติแบบโต้ตอบ (ตัวแก้ไขโค้ดในตัวและติวเตอร์ AI ตลอด 24/7) และปลดล็อคส่วนที่เหลือของคอร์ส RabbitMQ Messaging & Async Systems ให้อัปเกรดเป็น CoddyKit PRO คอร์ส RabbitMQ Messaging & Async Systems มีบทเรียนทั้งหมด 4 บทเรียน

คุณจะเรียนรู้อะไรในบทเรียน “การตอบรับจากคอนซูเมอร์และการจัดเข้าคิวใหม่”

ทำความเข้าใจการตอบรับจากคอนซูเมอร์ และเรียนรู้วิธีจัดข้อความกลับเข้าคิวเมื่อการประมวลผลล้มเหลว ออกแบบคอนซูเมอร์ที่ทนต่อข้อผิดพลาดและจัดการข้อผิดพลาดได้อย่างเหมาะสม คุณปฏิบัติ RabbitMQ Messaging & Async Systems ด้วยโค้ดที่ใช้งานได้จริงที่คุณเรียกใช้โดยตรงในเบราว์เซอร์ และติวเตอร์ AI ตลอด 24/7 ตอบคำถามของคุณขณะที่คุณไปผ่านบทเรียน

คุณต้องมีประสบการณ์ก่อนที่จะเริ่มเรียน RabbitMQ Messaging & Async Systems หรือไม่

ไม่จำเป็นต้องมีประสบการณ์มาก่อน RabbitMQ Messaging & Async Systems บน CoddyKit ออกแบบมาสำหรับผู้เริ่มต้นไปจนถึงผู้เรียนขั้นสูง คุณสามารถเริ่มต้นที่นี่หรือเริ่มจากตัวแรกและเรียนด้วยความเร็วของคุณเอง นี่คือบทเรียนที่ 3 จากทั้งหมด 4 บทเรียน

บทเรียน “การตอบรับจากคอนซูเมอร์และการจัดเข้าคิวใหม่” ใช้เวลานานแค่ไหน

บทเรียน CoddyKit ส่วนใหญ่ใช้เวลาประมาณ 5–10 นาที แต่ละบทเรียนจึงสั้นและเป็นแบบโต้ตอบ คุณสามารถก้าวหน้าอย่างต่อเนื่องและกลับมาเรียนต่อจากตรงที่เพิ่งหยุดบนเว็บและแอปได้เลย

ฉันเขียนและรันโค้ดในบทเรียน RabbitMQ Messaging & Async Systems นี้ได้ไหม

ได้ บทเรียน RabbitMQ Messaging & Async Systems ทุกบทมีตัวแก้ไขโค้ดในตัว คุณจึงเขียนและรันโค้ดจริงได้เลยในเบราว์เซอร์ และได้รับข้อเสนอแนะจาก AI ในทันที — ไม่ต้องติดตั้งในเครื่องของคุณ

บทเรียนทั้งหมดในหลักสูตรนี้

ข้อความและคิวแบบคงอยู่
การยืนยันจากผู้เผยแพร่เพื่อความน่าเชื่อถือ
การตอบรับจากคอนซูเมอร์และการจัดเข้าคิวใหม่
ธุรกรรมกับการยืนยันจากผู้เผยแพร่

← กลับไปที่ RabbitMQ Messaging & Async Systems