菜单
Message queue systems for game servers including Kafka, RabbitMQ, and actor models
默认使用会先检查来源的 Prompt；你也可以切换为直接命令，或下载本地副本。
决定是否安装前，请先阅读 SKILL.md，以及 SkillsMP 当前展示的配套文件。
用 Codex 或 Claude 帮你安装 复制这段 Prompt，粘贴到 Codex、Claude 或其他助手里，让它检查 Skill 页面并帮你完成安装。
直接命令不会经过审查 Prompt；运行前请先检查来源。
npx skills add https://github.com/pluginagentmarketplace/custom-plugin-server-side-game-dev --skill message-queues命令会保持在同一行。复制前请横向滚动并检查完整内容。
想先保存到本地？可下载 SkillsMP 当前能够提供的文件。
Asynchronous programming models including coroutines, async/await, and reactive patterns
Game server communication protocols including gRPC, REST, and custom binary protocols
Efficient data serialization for game networking including Protobuf, FlatBuffers, and custom binary
基于 SOC 职业分类
正在显示 SKILL.md
| name | message-queues |
| description | Message queue systems for game servers including Kafka, RabbitMQ, and actor models |
| sasmp_version | 1.3.0 |
| version | 2.0.0 |
| bonded_agent | 01-game-server-architect |
| bond_type | SECONDARY_BOND |
| parameters | {"required":["queue_system"],"optional":["batch_size","ack_mode"],"validation":{"queue_system":{"type":"string","enum":["kafka","rabbitmq","redis_pubsub","nats","sqs"]},"batch_size":{"type":"integer","min":1,"max":1000,"default":100},"ack_mode":{"type":"string","enum":["auto","manual","batch"],"default":"manual"}}} |
| retry_config | {"max_attempts":5,"backoff":"exponential","initial_delay_ms":100,"max_delay_ms":30000,"retryable_errors":["CONNECTION_LOST","BROKER_UNAVAILABLE"]} |
| observability | {"logging":{"level":"info","fields":["queue","topic","partition","offset"]},"metrics":[{"name":"messages_published_total","type":"counter"},{"name":"messages_consumed_total","type":"counter"},{"name":"consumer_lag","type":"gauge"},{"name":"processing_duration_ms","type":"histogram"}]} |
Implement asynchronous messaging for scalable game server architecture.
| System | Throughput | Latency | Ordering | Use Case |
|---|---|---|---|---|
| Kafka | Very High | Medium | Partition | Analytics, events |
| RabbitMQ | High | Low | Queue | Game events |
| Redis Pub/Sub | Very High | Very Low | None | Real-time updates |
| NATS | Very High | Ultra Low | Stream | Game state sync |
| SQS | High | Medium | FIFO option | Cloud native |
// Producer configuration
Properties producerProps = new Properties();
producerProps.put("bootstrap.servers", "kafka:9092");
producerProps.put("key.serializer", StringSerializer.class.getName());
producerProps.put("value.serializer", JsonSerializer.class.getName());
producerProps.put("acks", "all"); // Durability
producerProps.put("retries", 3);
producerProps.put("linger.ms", 5); // Batch for throughput
producerProps.put("batch.size", 16384);
KafkaProducer<String, GameEvent> producer = new KafkaProducer<>(producerProps);
// Publish game events
public {
ProducerRecord<String, GameEvent> record = <>(
,
event.getPlayerId(),
event
);
producer.send(record, (metadata, exception) -> {
(exception != ) {
log.error(, exception);
}
});
}
();
consumerProps.put(, );
consumerProps.put(, );
consumerProps.put(, );
consumerProps.put(, );
KafkaConsumer<String, GameEvent> consumer = <>(consumerProps);
consumer.subscribe(List.of());
(running) {
ConsumerRecords<String, GameEvent> records = consumer.poll(Duration.ofMillis());
(ConsumerRecord<String, GameEvent> record : records) {
processEvent(record.value());
}
consumer.commitSync();
}
// Connection with retry
func connectRabbitMQ() (*amqp.Connection, error) {
var conn *amqp.Connection
var err error
for i := 0; i < 5; i++ {
conn, err = amqp.Dial("amqp://guest:guest@localhost:5672/")
if err == nil {
return conn, nil
}
time.Sleep(time.Second * time.Duration(1<<i)) // Exponential backoff
}
return nil, fmt.Errorf("failed to connect after retries: %w", err)
}
// Publisher
func publishMatchEvent(ch *amqp.Channel, event MatchEvent) error {
body, err := json.Marshal(event)
if err != nil {
return err
}
return ch.Publish(
"game-exchange", // exchange
"match.created", // routing key
false, // mandatory
false, // immediate
amqp.Publishing{
ContentType: "application/json",
Body: body,
DeliveryMode: amqp.Persistent, // Survive broker restart
MessageId: uuid.New().String(),
Timestamp: time.Now(),
},
)
}
// Consumer with manual ack
func consumeMatchEvents(ch *amqp.Channel) error {
msgs, err := ch.Consume(
"match-events", // queue
"", // consumer tag
false, // auto-ack (false = manual)
false, // exclusive
false, // no-local
false, // no-wait
nil, // args
)
if err != nil {
return err
}
for msg := range msgs {
var event MatchEvent
if err := json.Unmarshal(msg.Body, &event); err != nil {
msg.Nack(false, false) // Don't requeue malformed messages
continue
}
if err := processMatchEvent(event); err != nil {
msg.Nack(false, true) // Requeue for retry
continue
}
msg.Ack(false) // Acknowledge successful processing
}
return nil
}
import redis
import json
from concurrent.futures import ThreadPoolExecutor
# Publisher (Game Server)
class GameStatePublisher:
def __init__(self):
self.redis = redis.Redis(host='localhost', port=6379)
def broadcast_state(self, game_id: str, state: dict):
channel = f"game:{game_id}"
self.redis.publish(channel, json.dumps(state))
def broadcast_chat(self, game_id: str, message: dict):
channel = f"chat:{game_id}"
self.redis.publish(channel, json.dumps(message))
# Subscriber (Client Gateway)
class GameStateSubscriber:
def __init__(self, game_id: str, callback):
self.redis = redis.Redis(host='localhost', port=6379)
self.pubsub = self.redis.pubsub()
self.callback = callback
self.game_id = game_id
def subscribe(self):
self.pubsub.subscribe(f"game:{self.game_id}")
for message in self.pubsub.listen():
if message['type'] == 'message':
data = json.loads(message['data'])
self.callback(data)
def unsubscribe(self):
self.pubsub.unsubscribe()
self.pubsub.close()
# Usage with connection pool
pool = redis.ConnectionPool(host='localhost', port=6379, max_connections=100)
redis_client = redis.Redis(connection_pool=pool)
// NATS JetStream for persistent messaging
func setupNATS() (*nats.Conn, nats.JetStreamContext, error) {
nc, err := nats.Connect("nats://localhost:4222",
nats.RetryOnFailedConnect(true),
nats.MaxReconnects(10),
nats.ReconnectWait(time.Second),
)
if err != nil {
return nil, nil, err
}
js, err := nc.JetStream()
if err != nil {
return nil, nil, err
}
// Create stream for game events
_, err = js.AddStream(&nats.StreamConfig{
Name: "GAME_EVENTS",
Subjects: []string{"game.>"},
Retention: nats.LimitsPolicy,
MaxAge: time.Hour * 24,
Storage: nats.FileStorage,
Replicas: 3,
})
return nc, js, err
}
// Publish with acknowledgment
func publishGameEvent(js nats.JetStreamContext, event GameEvent) error {
data, _ := json.Marshal(event)
ack, err := js.Publish(
fmt.Sprintf("game.%s.%s", event.GameID, event.Type),
data,
)
if err != nil {
return err
}
log.Printf("Published: seq=%d", ack.Sequence)
return nil
}
// Durable consumer
func consumeGameEvents(js nats.JetStreamContext) error {
sub, err := js.Subscribe("game.>",
func(msg *nats.Msg) {
var event GameEvent
json.Unmarshal(msg.Data, &event)
processEvent(event)
msg.Ack()
},
nats.Durable("game-processor"),
nats.ManualAck(),
nats.AckWait(time.Second*30),
)
if err != nil {
return err
}
defer sub.Unsubscribe()
<-make(chan struct{}) // Block forever
return nil
}
// Orleans Grain (Virtual Actor)
public interface IPlayerGrain : IGrainWithStringKey
{
Task<PlayerState> GetState();
Task<bool> TakeDamage(int amount, string sourceId);
Task<bool> ApplyBuff(Buff buff);
}
public class PlayerGrain : Grain, IPlayerGrain
{
private readonly IPersistentState<PlayerState> _state;
private readonly ILogger<PlayerGrain> _logger;
public PlayerGrain(
[PersistentState("player", "gameStore")] IPersistentState<PlayerState> state,
ILogger<PlayerGrain> logger)
{
_state = state;
_logger = logger;
}
public Task<PlayerState> GetState() => Task.FromResult(_state.State);
public async Task<bool> TakeDamage(int amount, string sourceId)
{
_state.State.Health -= amount;
if (_state.State.Health <= 0)
{
// Notify game grain about death
var gameGrain = GrainFactory.GetGrain<IGameGrain>(_state.State.GameId);
await gameGrain.OnPlayerDeath(this.GetPrimaryKeyString(), sourceId);
}
await _state.WriteStateAsync();
return _state.State.Health > 0;
}
}
// Silo configuration
var host = new HostBuilder()
.UseOrleans(siloBuilder =>
{
siloBuilder
.UseLocalhostClustering()
.AddRedisGrainStorage("gameStore", options =>
{
options.ConnectionString = "localhost:6379";
})
.ConfigureLogging(logging => logging.AddConsole());
})
.Build();
| Use Case | Recommended | Reason |
|---|---|---|
| Cross-server chat | RabbitMQ | Reliable delivery |
| Analytics pipeline | Kafka | High throughput, replay |
| Real-time state | Redis Pub/Sub | Ultra-low latency |
| Distributed game state | Orleans/Akka | Location transparency |
| Match results | Kafka | Ordered, durable |
| Notifications | NATS | Simple, fast |
| Error | Root Cause | Solution |
|---|---|---|
| Consumer lag | Slow processing | Scale consumers |
| Message loss | Auto-ack before process | Manual ack |
| Duplicate processing | At-least-once | Idempotent handlers |
| Broker unavailable | Single point | Cluster mode |
# Kafka consumer lag
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
--describe --group analytics-consumer
# RabbitMQ queue status
rabbitmqctl list_queues name messages consumers
# Redis pub/sub channels
redis-cli PUBSUB CHANNELS "game:*"
# NATS stream info
nats stream info GAME_EVENTS
func TestMessagePublishing(t *testing.T) {
// Setup test broker (use testcontainers)
container := setupRabbitMQContainer(t)
defer container.Terminate(context.Background())
conn, _ := amqp.Dial(container.URI)
ch, _ := conn.Channel()
// Publish test event
event := MatchEvent{
MatchID: "match-123",
EventType: "created",
}
err := publishMatchEvent(ch, event)
require.NoError(t, err)
// Verify message received
msgs, _ := ch.Consume("match-events", "", true, false, false, false, nil)
select {
case msg := <-msgs:
var received MatchEvent
json.Unmarshal(msg.Body, &received)
assert.Equal(t, event.MatchID, received.MatchID)
case <-time.After(time.Second * 5):
t.Fatal("timeout waiting for message")
}
}
assets/ - Queue configurationsreferences/ - Messaging patterns