Message Queues for Game Servers
Implement asynchronous messaging for scalable game server architecture.
Queue Systems Comparison
| 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 |
Apache Kafka for Game Analytics
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");
producerProps.put("retries", 3);
producerProps.put("linger.ms", 5);
producerProps.put("batch.size", 16384);
KafkaProducer<String, GameEvent> producer = new KafkaProducer<>(producerProps);
public void publishEvent(GameEvent event) {
ProducerRecord<String, GameEvent> record = new ProducerRecord<>(
"game-events",
event.getPlayerId(),
event
);
producer.send(record, (metadata, exception) -> {
if (exception != null) {
log.error("Failed to publish event", exception);
}
});
}
Properties consumerProps = new Properties();
consumerProps.put("bootstrap.servers", "kafka:9092");
consumerProps.put("group.id", "analytics-consumer");
consumerProps.put("auto.offset.reset", "earliest");
consumerProps.put("enable.auto.commit", false);
KafkaConsumer<String, GameEvent> consumer = new KafkaConsumer<>(consumerProps);
consumer.subscribe(List.of("game-events"));
while (running) {
ConsumerRecords<String, GameEvent> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, GameEvent> record : records) {
processEvent(record.value());
}
consumer.commitSync();
}
RabbitMQ for Game Commands
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))
}
return nil, fmt.Errorf("failed to connect after retries: %w", err)
}
func publishMatchEvent(ch *amqp.Channel, event MatchEvent) error {
body, err := json.Marshal(event)
if err != nil {
return err
}
return ch.Publish(
"game-exchange",
"match.created",
false,
false,
amqp.Publishing{
ContentType: "application/json",
Body: body,
DeliveryMode: amqp.Persistent,
MessageId: uuid.New().String(),
Timestamp: time.Now(),
},
)
}
{
msgs, err := ch.Consume(
,
,
,
,
,
,
,
)
err != {
err
}
msg := msgs {
event MatchEvent
err := json.Unmarshal(msg.Body, &event); err != {
msg.Nack(, )
}
err := processMatchEvent(event); err != {
msg.Nack(, )
}
msg.Ack()
}
}
Redis Pub/Sub for Real-Time
import redis
import json
from concurrent.futures import ThreadPoolExecutor
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))
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()
message .pubsub.listen():
message[] == :
data = json.loads(message[])
.callback(data)
():
.pubsub.unsubscribe()
.pubsub.close()
pool = redis.ConnectionPool(host=, port=, max_connections=)
redis_client = redis.Redis(connection_pool=pool)
NATS for Low-Latency 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
}
_, 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
}
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
}
func consumeGameEvents {
sub, err := js.Subscribe(,
{
event GameEvent
json.Unmarshal(msg.Data, &event)
processEvent(event)
msg.Ack()
},
nats.Durable(),
nats.ManualAck(),
nats.AckWait(time.Second*),
)
err != {
err
}
sub.Unsubscribe()
<-( {})
}
Actor Model (Akka/Orleans)
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)
{
var gameGrain = GrainFactory.GetGrain<IGameGrain>(_state.State.GameId);
await gameGrain.OnPlayerDeath(.GetPrimaryKeyString(), sourceId);
}
_state.WriteStateAsync();
_state.State.Health > ;
}
}
host = HostBuilder()
.UseOrleans(siloBuilder =>
{
siloBuilder
.UseLocalhostClustering()
.AddRedisGrainStorage(, options =>
{
options.ConnectionString = ;
})
.ConfigureLogging(logging => logging.AddConsole());
})
.Build();
Use Case Mapping
| 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 |
Troubleshooting
Common Failure Modes
| 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 |
Debug Checklist
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
--describe --group analytics-consumer
rabbitmqctl list_queues name messages consumers
redis-cli PUBSUB CHANNELS "game:*"
nats stream info GAME_EVENTS
Unit Test Template
func TestMessagePublishing(t *testing.T) {
container := setupRabbitMQContainer(t)
defer container.Terminate(context.Background())
conn, _ := amqp.Dial(container.URI)
ch, _ := conn.Channel()
event := MatchEvent{
MatchID: "match-123",
EventType: "created",
}
err := publishMatchEvent(ch, event)
require.NoError(t, err)
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")
}
}
Resources
assets/ - Queue configurationsreferences/ - Messaging patterns
