Skip to content

Commit 8efc28d

Browse files
author
“Kevin”
committed
Add blog for batch future
1 parent c232199 commit 8efc28d

File tree

1 file changed

+229
-0
lines changed

1 file changed

+229
-0
lines changed
Lines changed: 229 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,229 @@
1+
---
2+
title: "Introducing Batch Future: Controlled Concurrency for Activity Execution"
3+
description: "We're excited to announce Batch Future, a new feature in the Cadence Go client that provides controlled concurrency for bulk operations, preventing overwhelming downstream services while maintaining efficient parallel processing."
4+
date: 2025-09-19
5+
authors: kevinb
6+
tags:
7+
- announcement
8+
- performance
9+
- releases
10+
---
11+
12+
Are you struggling with uncontrolled concurrency when processing multiple activities? Do you find yourself hitting rate limits or overwhelming downstream services when running bulk operations? We've got great news for you!
13+
14+
Today, we're thrilled to announce **Batch Future**—a powerful new feature in the Cadence Go client that provides controlled concurrency for bulk operations. You can now process multiple activities in parallel while maintaining precise control over how many run simultaneously.
15+
16+
<!-- truncate -->
17+
18+
## The Problem: Uncontrolled Concurrency
19+
20+
Traditionally, when you need to process multiple items in a Cadence workflow, you'd write something like this:
21+
22+
```go
23+
func ProcessUsers(ctx workflow.Context, userIDs []string) error {
24+
var futures []workflow.Future
25+
for _, userID := range userIDs {
26+
future := workflow.ExecuteActivity(ctx, UpdateUserActivity, userID)
27+
futures = append(futures, future)
28+
}
29+
30+
// Wait for all activities to complete
31+
for _, future := range futures {
32+
if err := future.Get(ctx, nil); err != nil {
33+
return err
34+
}
35+
}
36+
return nil
37+
}
38+
```
39+
40+
This approach works, but it has **uncontrolled concurrency**:
41+
- All activities start simultaneously, potentially overwhelming downstream services
42+
- No way to limit concurrent executions
43+
- Difficult to manage resource usage
44+
- Can cause rate limiting or timeouts
45+
46+
## The Solution: Batch Future
47+
48+
With Batch Future, you can process users with **controlled concurrency**:
49+
50+
```go
51+
func ProcessUsersBatch(ctx workflow.Context, userIDs []string, concurrency int) error {
52+
// Create activity factories for each user
53+
factories := make([]func(workflow.Context) workflow.Future, len(userIDs))
54+
for i, userID := range userIDs {
55+
userID := userID // Capture loop variable for closure
56+
factories[i] = func(ctx workflow.Context) workflow.Future {
57+
return workflow.ExecuteActivity(ctx, UpdateUserActivity, userID)
58+
}
59+
}
60+
61+
// Execute with controlled concurrency
62+
batch, err := workflow.NewBatchFuture(ctx, concurrency, factories)
63+
if err != nil {
64+
return fmt.Errorf("failed to create batch future: %w", err)
65+
}
66+
67+
// Wait for all activities to complete
68+
return batch.Get(ctx, nil)
69+
}
70+
```
71+
72+
## Key Benefits: Controlled Concurrency
73+
74+
Batch Future provides several important advantages:
75+
76+
- **Controlled Concurrency**: Limit simultaneous executions to prevent overwhelming downstream services
77+
- **Resource Management**: Better control over memory and CPU usage
78+
- **Rate Limiting Protection**: Avoid hitting API rate limits by controlling execution speed
79+
- **Graceful Cancellation**: All activities can be cancelled together if needed
80+
- **Simplified Error Handling**: Single point of failure handling for the entire batch
81+
82+
## Real-World Use Cases
83+
84+
Batch Future is perfect for scenarios like:
85+
86+
### 1. Multi-Service Data Synchronization
87+
```go
88+
func SyncProductData(ctx workflow.Context, products []Product) error {
89+
// Sync to multiple services with different concurrency limits
90+
inventoryBatch := createBatch(ctx, products, 5, SyncToInventoryActivity)
91+
searchBatch := createBatch(ctx, products, 3, SyncToSearchActivity)
92+
analyticsBatch := createBatch(ctx, products, 2, SyncToAnalyticsActivity)
93+
94+
// Wait for all sync operations to complete
95+
if err := inventoryBatch.Get(ctx, nil); err != nil {
96+
return fmt.Errorf("inventory sync failed: %w", err)
97+
}
98+
if err := searchBatch.Get(ctx, nil); err != nil {
99+
return fmt.Errorf("search sync failed: %w", err)
100+
}
101+
return analyticsBatch.Get(ctx, nil)
102+
}
103+
104+
func createBatch(ctx workflow.Context, items []Product, concurrency int, activityFunc interface{}) workflow.Future {
105+
factories := make([]func(workflow.Context) workflow.Future, len(items))
106+
for i, item := range items {
107+
item := item
108+
factories[i] = func(ctx workflow.Context) workflow.Future {
109+
return workflow.ExecuteActivity(ctx, activityFunc, item)
110+
}
111+
}
112+
batch, _ := workflow.NewBatchFuture(ctx, concurrency, factories)
113+
return batch
114+
}
115+
```
116+
117+
### 2. Progressive Data Processing with Different Priorities
118+
```go
119+
func ProcessDataWithPriorities(ctx workflow.Context, data []DataItem) error {
120+
// High priority items get more concurrency
121+
highPriority := filterByPriority(data, "high")
122+
lowPriority := filterByPriority(data, "low")
123+
124+
// Process high priority items first with high concurrency
125+
highBatch, _ := workflow.NewBatchFuture(ctx, 10, createFactories(highPriority, ProcessHighPriorityActivity))
126+
127+
// Wait for high priority to complete, then process low priority with lower concurrency
128+
if err := highBatch.Get(ctx, nil); err != nil {
129+
return err
130+
}
131+
132+
lowBatch, _ := workflow.NewBatchFuture(ctx, 3, createFactories(lowPriority, ProcessLowPriorityActivity))
133+
return lowBatch.Get(ctx, nil)
134+
}
135+
```
136+
137+
### 3. Conditional Batch Processing with Retry Logic
138+
```go
139+
func ProcessOrdersWithRetry(ctx workflow.Context, orders []Order) error {
140+
// First attempt with normal concurrency
141+
factories := make([]func(workflow.Context) workflow.Future, len(orders))
142+
for i, order := range orders {
143+
order := order
144+
factories[i] = func(ctx workflow.Context) workflow.Future {
145+
return workflow.ExecuteActivity(ctx, ProcessOrderActivity, order)
146+
}
147+
}
148+
149+
batch, _ := workflow.NewBatchFuture(ctx, 5, factories)
150+
if err := batch.Get(ctx, nil); err != nil {
151+
// If batch fails, retry failed orders individually with higher concurrency
152+
return retryFailedOrders(ctx, orders, 10)
153+
}
154+
return nil
155+
}
156+
```
157+
158+
159+
## How It Works Under the Hood
160+
161+
Batch Future leverages Cadence's existing activity infrastructure with controlled concurrency:
162+
163+
1. **Activity Factories**: Creates lazy-evaluated activity functions that aren't executed until needed
164+
2. **Concurrency Control**: Limits the number of activities running simultaneously
165+
3. **Queue Management**: Maintains a queue of pending activities and starts new ones as others complete
166+
4. **Future Management**: Returns a single future that completes when all activities finish
167+
5. **Error Propagation**: If any activity fails, the entire batch can be cancelled or retried
168+
169+
## Getting Started
170+
171+
Ready to supercharge your workflows? Here's how to get started:
172+
173+
### 1. Update Your Go Client
174+
Make sure you're using the latest version of the Cadence Go client:
175+
176+
```bash
177+
go get github.com/uber/cadence-go-client@latest
178+
```
179+
180+
### 2. Try the Sample
181+
Check out our [Batch Future sample](https://github.com/cadence-workflow/cadence-samples/tree/master/cmd/samples/batch) to see it in action.
182+
183+
### 3. Migrate Your Workflows
184+
Identify workflows that process multiple items and convert them to use Batch Future:
185+
186+
```go
187+
// Before: Uncontrolled concurrency
188+
var futures []workflow.Future
189+
for _, item := range items {
190+
future := workflow.ExecuteActivity(ctx, ProcessItem, item)
191+
futures = append(futures, future)
192+
}
193+
for _, future := range futures {
194+
if err := future.Get(ctx, nil); err != nil {
195+
return err
196+
}
197+
}
198+
199+
// After: Controlled concurrency with Batch Future
200+
factories := make([]func(workflow.Context) workflow.Future, len(items))
201+
for i, item := range items {
202+
item := item // Capture loop variable
203+
factories[i] = func(ctx workflow.Context) workflow.Future {
204+
return workflow.ExecuteActivity(ctx, ProcessItem, item)
205+
}
206+
}
207+
batch, err := workflow.NewBatchFuture(ctx, 3, factories) // Limit to 3 concurrent
208+
if err != nil {
209+
return err
210+
}
211+
return batch.Get(ctx, nil)
212+
```
213+
214+
## Best Practices
215+
216+
- **Choose Appropriate Concurrency**: Start with 3-5 concurrent activities and adjust based on downstream service capacity
217+
- **Activity Factories**: Always capture loop variables in closures to avoid race conditions
218+
- **Error Handling**: Implement proper error handling for individual activity failures
219+
- **Resource Management**: Consider memory usage for large batches
220+
- **Monitoring**: Use heartbeats for long-running activities within the batch
221+
222+
223+
## Try It Today!
224+
225+
Batch Future is available now in the latest Cadence Go client. We can't wait to see how you use it to optimize your workflows!
226+
227+
Have questions or feedback? Join our [Slack community](http://t.uber.com/cadence-slack) or open an issue on [GitHub](https://github.com/cadence-workflow/cadence-go-client).
228+
229+
Happy coding, and here's to faster, more efficient workflows!

0 commit comments

Comments
 (0)