YMJake.RocketMQ 0.1.0-preview.2

YMJake.RocketMQ

English | 简体中文

An unofficial Apache RocketMQ client for .NET, speaking the RocketMQ 5.x gRPC proxy protocol.

It ships net8.0 and net10.0 targets and exposes typed, task-based APIs for the Producer, SimpleConsumer, PushConsumer, and the lite-topic consumers. The client half of the protocol is implemented from scratch: settings registration, the telemetry stream, queue assignment, long polling, receipt handles, message-body integrity checks, and reconnect.

Preview. The package is 0.1.0-preview.1. The public API may still change before 1.0.

Requirements

• .NET 8 SDK or newer; the library targets net8.0 and net10.0.
• A RocketMQ 5.x deployment with the proxy enabled (gRPC, default port 8081). The client connects to the proxy access point, not to a NameServer or Broker address.
• Lite topics need a lite-enabled deployment; LiteSimpleConsumer additionally needs RocketMQ 5.5.1 or later, because older proxies reject LITE_SIMPLE_CONSUMER.

Install

dotnet add package YMJake.RocketMQ --prerelease

The package ID and the namespace differ; the code is imported with:

using RocketMQ.Client;

Features and status

✅ means the client implements the API. The corresponding server-side capability and configuration still apply.

Intentionally not included: the classic Remoting protocol, an admin API, Microsoft dependency injection / Generic Host integration, and OpenTelemetry metrics or tracing.

Quick start

Producer

using RocketMQ.Client;

var config = new ClientConfig { Endpoints = "localhost:8081" };

// Topics are pre-warmed at create, so a topic missing on the server fails the start;
// without Topics the route is fetched on the first send.
var options = new ProducerOptions { MaxAttempts = 3 };
options.Topics.Add("TopicTest");

await using var producer = await Producer.Create(config, options);
var receipt = await producer.SendAsync(new Message("TopicTest", "hello"));

Push consumer

using System.Text;
using RocketMQ.Client;

var config = new ClientConfig { Endpoints = "localhost:8081" };

var options = new PushConsumerOptions { ConsumerGroup = "my-consumer-group" };
options.SubscriptionExpressions["TopicTest"] = FilterExpression.All;

await using var consumer = await PushConsumer.Create(config, options, view =>
{
    Console.WriteLine(Encoding.UTF8.GetString(view.Body));
    return ConsumeResult.Success;
});

Simple consumer

using RocketMQ.Client;

var config = new ClientConfig { Endpoints = "localhost:8081" };

var options = new SimpleConsumerOptions
{
    ConsumerGroup = "my-consumer-group",
    AwaitDuration = TimeSpan.FromSeconds(30),
};
options.SubscriptionExpressions["TopicTest"] = FilterExpression.All;

await using var consumer = await SimpleConsumer.Create(config, options);

while (true)
{
    var messages = await consumer.ReceiveAsync("TopicTest", 16, TimeSpan.FromSeconds(15));
    foreach (var message in messages)
    {
        await consumer.AckAsync(message);
    }
}

Lite consumers

using System.Text;
using RocketMQ.Client;

var config = new ClientConfig { Endpoints = "localhost:8081" };

// One bind topic carries all lite topics; the client subscribes it with "*" unless you do.
var options = new PushConsumerOptions { ConsumerGroup = "my-lite-group" };

await using var consumer = await LitePushConsumer.Create(config, "TopicTest", options, view =>
{
    Console.WriteLine($"{view.LiteTopic}: {Encoding.UTF8.GetString(view.Body)}");
    return ConsumeResult.Success;
});

await consumer.SubscribeLiteAsync("order-1", OffsetOption.Min);

Publish with new Message("TopicTest", "hello") { LiteTopic = "order-1" }.

Lite consumers require the server-side lite capability: the bind topic must accept lite messages (message.type=LITE), and the consumer group must be a lite group (lite.bind.topic=<bind topic>).

Credentials

var config = new ClientConfig
{
    Endpoints = "localhost:8081",
    CredentialsProvider = new StaticCredentialsProvider(accessKey, accessSecret),
};

Delivery semantics

• Push: returning ConsumeResult.Success acks the message. Returning ConsumeResult.Failure, or throwing from the listener, leaves it unacked; the server redelivers it after the retry policy's backoff. A FIFO message is redelivered locally, keeping its receipt handle and advancing MessageView.DeliveryAttempt; after MaxAttempts it is forwarded to the dead-letter queue.
• Simple: a received message stays invisible for the requested duration; ack it, or extend the duration with ChangeInvisibleDurationAsync before the receipt handle expires.
• Ordering: FIFO ordering is decided by the consumer group's configuration on the server (isConsumeMessageOrderly), which the client reads during the settings handshake.

Compatibility

The gRPC push consumer is client-initiated long polling, not protocol-level Broker push. SQL filtering requires the target server's filtering configuration. This is an unofficial project and is not affiliated with or endorsed by the Apache Software Foundation.

License

Licensed under the Apache License 2.0.