Building a Pub/Sub Service in Go

Understanding the Pub/Sub Pattern

The publish-subscribe (Pub/Sub) pattern is a messaging architecture that enables asynchronous communication between components in distributed systems. By decoupling publishers from subscribers through a message broker, this pattern powers everything from real-time chat applications to event-driven microservices.

According to Google Cloud's Pub/Sub documentation, this pattern allows for loose coupling between senders and receivers, making it ideal for building scalable, resilient architectures that can adapt to changing requirements without extensive refactoring.

Core Components

• Publishers: Send messages to specific topics without knowledge of subscribers
• Subscribers: Express interest in topics and receive relevant messages
• Message Broker: Manages message routing between publishers and subscribers
• Topics: Named channels that categorize and route messages

When to Use Pub/Sub

Pub/Sub is ideal for scenarios requiring asynchronous messaging and decoupled communication:

• Microservices Architecture: Enables services to communicate without direct dependencies
• Real-Time Data Processing: Handles streaming data and immediate event reactions
• Event-Driven Architectures: Facilitates flow of events within a system
• Internet of Things (IoT): Enables devices to publish data and other components to consume it
• Distributed Systems: Provides loose coupling and fault-tolerant communication

For applications built with our web development services, implementing Pub/Sub patterns can significantly improve system scalability and maintainability.

Building Pub/Sub with Go Channels and Goroutines

Go's native concurrency primitives make implementing Pub/Sub patterns straightforward. The fundamental approach uses channels for communication and goroutines for concurrent message processing, as demonstrated in LogRocket's comprehensive guide to building pub/sub services in Go.

Basic Implementation Steps

1. Define a channel for publisher-subscriber communication
2. Create publisher goroutines that send messages to channels
3. Create subscriber goroutines that receive and process messages
4. Use sync.WaitGroup for goroutine synchronization

Code Example: Basic Publisher and Subscriber

func publisher(wg *sync.WaitGroup, msgChan chan string) {
 for i := 0; i < 10; i++ {
 msgChan <- fmt.Sprintf("Message %d", i)
 }
 close(msgChan)
 wg.Done()
}

func subscriber(id int, wg *sync.WaitGroup, msgChan chan string) {
 for message := range msgChan {
 fmt.Printf("Subscriber %d received: %s\n", id, message)
 }
 wg.Done()
}

func main() {
 msgChan := make(chan string)
 wg := &sync.WaitGroup{}
 
 wg.Add(1)
 go publisher(wg, msgChan)
 
 for i := 0; i < 3; i++ {
 wg.Add(1)
 go subscriber(i, wg, msgChan)
 }
 
 wg.Wait()
}

This basic implementation demonstrates the core concepts: publishers send messages through channels, subscribers receive them asynchronously, and the WaitGroup ensures all goroutines complete before the program exits.

Advanced Implementation: Multi-Topic Pub/Sub Service

For production systems, implement a proper interface-based design with thread-safe subscription management. According to implementation guides from GoScrapy, a robust Pub/Sub service should support multiple topics, concurrent access, and proper synchronization.

type PubSub interface {
 Publish(topic string, message interface{})
 Subscribe(topic string) <-chan interface{}
 Wait()
}

type PubSubImpl struct {
 waitGroup sync.WaitGroup
 topics map[string][]chan interface{}
 subscriptionLock sync.Mutex
}

func (ps *PubSubImpl) Publish(topic string, message interface{}) {
 ps.subscriptionLock.Lock()
 defer ps.subscriptionLock.Unlock()
 
 subscribers := ps.topics[topic]
 for _, subscriber := range subscribers {
 ps.waitGroup.Add(1)
 go func(sub chan interface{}) {
 sub <- message
 ps.waitGroup.Done()
 }(subscriber)
 }
}

func (ps *PubSubImpl) Subscribe(topic string) <-chan interface{} {
 ps.subscriptionLock.Lock()
 defer ps.subscriptionLock.Unlock()
 
 subscriber := make(chan interface{})
 ps.topics[topic] = append(ps.topics[topic], subscriber)
 return subscriber
}

This implementation provides thread-safe operations using mutex locks, supports multiple topics through a map structure, and properly manages goroutine lifecycles with WaitGroups. For enterprise applications requiring advanced messaging capabilities, our custom web development services can help architect robust Pub/Sub solutions.

Using Redis for Pub/Sub

Redis provides a lightweight Pub/Sub implementation suitable for many production scenarios. It offers low-latency messaging with automatic reconnection on connection errors, making it an excellent choice for real-time applications. When implementing distributed messaging, consider how this integrates with your overall web application architecture.

Server Implementation

rdb := redis.NewClient(&redis.Options{
 Addr: "localhost:6379",
 Password: "",
 DB: 0,
})

ctx := context.Background()
pubsub := rdb.Subscribe(ctx, "channel-name")
defer pubsub.Close()
ch := pubsub.Channel()

go func(ch <-chan *redis.Message) {
 for msg := range ch {
 fmt.Println(msg.Channel, msg.Payload)
 }
}(ch)

// Publish messages
user := User{Name: "John", Email: "[email protected]"}
payload, _ := json.Marshal(user)
rdb.Publish(ctx, "channel-name", payload)

Redis Pub/Sub is ideal for scenarios where you need simple, fast message broadcasting across multiple services. However, note that Redis Pub/Sub is fire-and-forget--messages are not persisted, so subscribers must be connected to receive messages as they are published.

Implementing NATS with JetStream

NATS with JetStream provides enhanced capabilities including message persistence, streaming, and at-least-once delivery semantics. This makes it suitable for production systems that require reliable message delivery. Our AI automation services often leverage such messaging patterns for intelligent system integrations.

NATS Server Setup

// Connect to NATS
nc, err := nats.Connect(nats.DefaultURL)

// Create JetStream context
js, err := nc.JetStream(nats.PublishAsyncMaxPending(256))

// Create a persistent stream
js.AddStream(&nats.StreamConfig{
 Name: "ORDERS",
 Subjects: []string{"ORDERS.*"},
})

// Subscribe to messages
js.Subscribe("ORDERS.*", func(m *nats.Msg) {
 fmt.Printf("Received: %s\n", string(m.Data))
})

Publishing with NATS

for i := 0; i < 500; i++ {
 js.PublishAsync("ORDERS.scratch", []byte(fmt.Sprintf("Hello %d", i)))
}

select {
case <-js.PublishAsyncComplete():
 fmt.Println("All messages published")
case <-time.After(5 * time.Second):
 fmt.Println("Timeout waiting for publishes")
}

JetStream's async publishing allows you to send messages without blocking, with callbacks to handle completion or errors. This approach is particularly valuable for high-throughput systems where message rate matters more than immediate confirmation.

Real-World Use Cases

Building a Real-Time Chat Application

A complete chat application requires WebSocket connections for real-time bidirectional communication combined with Pub/Sub for broadcasting messages to all connected clients. This architectural pattern is common in modern applications built with our custom web development solutions:

app := fiber.New()
pubsub := NewPubSub()
subscribe := pubsub.Subscribe("chat")
connections := make(map[string]*websocket.Conn)

app.Get("/ws/:id", websocket.New(func(c *websocket.Conn) {
 id := uuid.New().String()
 connections[id] = c
 defer delete(connections, id)
 
 for {
 _, msg, _ := c.ReadMessage()
 var m Msg
 json.Unmarshal(msg, &m)
 m.Id = id
 pubsub.Publish("chat", m)
 }
}))

// Broadcast to all connected clients
go func() {
 for message := range subscribe {
 data, _ := json.Marshal(message)
 for _, c := range connections {
 c.WriteMessage(1, data)
 }
 }
}()

Notification Systems

Pub/Sub powers notification systems for:

• Push notifications to mobile devices: Route alerts through dedicated notification services
• Email notification queues: Batch and send emails asynchronously
• In-app notification feeds: Update user interfaces in real-time
• System alerts and monitoring: Trigger alerts based on system events

These patterns are essential for building scalable applications that can handle thousands of concurrent users without overwhelming backend services. Integrating SEO best practices ensures these real-time features remain discoverable and performant.

Performance Optimization and Best Practices

Optimization Guidelines

1. Message Size: Keep messages small to reduce latency and storage overhead
2. Concurrency Control: Balance publisher/subscriber count to prevent resource exhaustion
3. Message Batching: Group messages for fewer API calls and better throughput
4. Load Balancing: Distribute traffic across multiple Pub/Sub nodes
5. Caching Layer: Cache frequently accessed messages to reduce load
6. Monitoring: Track throughput, latency, and error rates continuously

Production Considerations

When deploying Pub/Sub services in production environments, consider these factors:

• Error Handling: Implement proper retry mechanisms with exponential backoff
• Connection Pooling: Use connection pools for external Pub/Sub services to reduce overhead
• Message Retention: Configure appropriate retention policies based on business requirements
• Dead Letter Queues: Implement DLQs for messages that fail processing after retries
• Authentication: Use proper authentication mechanisms for secure messaging

Monitoring Metrics

Track these metrics for optimal performance and reliability:

• Message throughput (messages/second)
• End-to-end latency from publish to delivery
• Error rates and failure patterns
• Subscription lag between message publish and subscriber receipt
• Connection pool utilization and health

For applications requiring enterprise-grade reliability, our microservices architecture services can help design and implement robust Pub/Sub solutions.

Conclusion

Building a Pub/Sub service in Go enables you to create decoupled, scalable systems that handle asynchronous communication between components. Start with Go channels for simple in-process implementations, then scale to Redis for lightweight distributed scenarios or NATS JetStream for production requirements with persistence needs.

Key Takeaways

• Start Simple: Use Go channels for in-process messaging between goroutines
• Scale Appropriately: Move to Redis for simple distributed Pub/Sub or NATS for advanced features
• Focus on Reliability: Implement proper error handling, retry mechanisms, and monitoring
• Design for Scale: Consider partitioning, batching, and consumer groups from the start

The Pub/Sub pattern provides a foundation for building modern event-driven architectures that power today's most demanding applications. Whether you're building real-time chat systems, notification platforms, or microservices communication layers, understanding these patterns is essential for scalable web application development.

Sources

1. GoScrapy Blog - How to Write a Pub Sub Service With Go
2. LogRocket Blog - Building a pub/sub service in Go
3. Google Cloud Pub/Sub Documentation