KeySharedProducer.java

import com.danubemessaging.client.DanubeClient;
import com.danubemessaging.client.DispatchStrategy;
import com.danubemessaging.client.Producer;

import java.util.Map;

/**
 * Key-Shared producer example: sends messages with routing keys so the broker
 * dispatches them to consumers based on consistent hashing of the key.
 *
 * Run KeySharedConsumer.java or KeySharedFilteredConsumer.java in a separate terminal.
 *
 * Prerequisites: Danube broker running on localhost:6650
 *   cd docker && docker compose up -d
 */
public class KeySharedProducer {

    private static final String BROKER_URL = System.getenv().getOrDefault("DANUBE_BROKER_URL", "http://127.0.0.1:6650");
    private static final String TOPIC = "/default/topic_key_shared";

    public static void main(String[] args) throws Exception {
        DanubeClient client = DanubeClient.builder()
                .serviceUrl(BROKER_URL)
                .build();

        Producer producer = client.newProducer()
                .withTopic(TOPIC)
                .withName("producer_key_shared")
                .withDispatchStrategy(DispatchStrategy.RELIABLE)
                .build();

        producer.create();
        System.out.println("Key-Shared producer created");

        // Send messages with different routing keys.
        // All messages with the same key are guaranteed to be delivered
        // to the same consumer, in order.
        String[] keys = {"payment", "shipping", "invoice", "payment", "shipping"};

        for (int i = 0; i < keys.length; i++) {
            String key = keys[i];
            String payload = String.format("Order event #%d for key=%s", i, key);
            long msgId = producer.sendWithKey(payload.getBytes(), Map.of(), key);
            System.out.printf("Sent message id=%d key=%s payload=%s%n", msgId, key, payload);
            Thread.sleep(500);
        }

        System.out.println("Done");
        client.close();
    }
}