# Worker Concurrency and Batch Pulling
Run several bunqueue jobs at once with concurrency, change it at runtime, and cut round-trips on high-volume queues with batch pulling and long polling.
Canonical: https://bunqueue.dev/guide/worker/concurrency/
---
import { Tabs, TabItem } from '@astrojs/starlight/components';
guide · worker
More at once, fewer round-trips.
Concurrency decides how many jobs a single worker runs in parallel. Batch pulling decides how many it fetches per request. Together they set the ceiling on what one process can do.
## Process jobs in parallel
```typescript
const worker = new Worker('my-queue', processor, {
embedded: true,
concurrency: 5, // Up to 5 jobs at once (default: 1)
});
```
```typescript
const worker = new Worker('my-queue', processor, {
embedded: false,
concurrency: 5, // Up to 5 jobs at once (default: 1)
});
```
```python
worker = Worker("my-queue", process, concurrency=5) # default: 4
```
```php
// The PHP worker is sequential by design: one job at a time.
// Scale out by running more worker processes instead:
// php worker.php & php worker.php &
$worker = new Worker('my-queue', $processor);
```
```go
worker := bunqueue.NewWorker("my-queue", processor, bunqueue.WorkerOptions{
Concurrency: 5, // Up to 5 jobs at once (default: 4)
})
```
```rust
let worker = Worker::new("my-queue", processor, WorkerOptions {
concurrency: 5, // Up to 5 jobs at once (default: 4)
..Default::default()
});
```
```elixir
worker = Bunqueue.Worker.new("my-queue", handler, concurrency: 5) # default: 1
```
You can change concurrency at runtime, without restarting:
```typescript
worker.concurrency = 10; // Scale up under load
worker.concurrency = 2; // Scale back down (minimum: 1)
```
```typescript
worker.concurrency = 10; // Scale up under load
worker.concurrency = 2; // Scale back down (minimum: 1)
```
The Python SDK fixes concurrency at construction. Drain and replace the worker
to change it safely:
```python
worker.close()
worker = Worker("my-queue", process, concurrency=10)
```
PHP workers are sequential. Change process-level parallelism by starting or
stopping worker processes under your process supervisor.
```bash
php worker.php &
php worker.php &
```
The Go SDK fixes concurrency at construction. Stop and replace the worker:
```go
worker.Stop()
worker.Close()
worker = bunqueue.NewWorker("my-queue", processor,
bunqueue.WorkerOptions{Concurrency: 10})
```
The Rust SDK fixes concurrency at construction. Stop and replace the worker:
```rust
worker.stop();
worker.close();
let worker = Worker::new("my-queue", processor, WorkerOptions {
concurrency: 10,
..Default::default()
});
```
The Elixir SDK fixes concurrency at construction. Stop and replace the worker:
```elixir
:ok = Bunqueue.Worker.stop(worker)
worker = Bunqueue.Worker.new("my-queue", handler, concurrency: 10)
```
*Both TypeScript packages support runtime concurrency changes. The other SDKs fix `concurrency` at construction.*
## Batch pulling
For high-volume queues, pull many jobs per round-trip and long-poll while idle:
```typescript
const worker = new Worker('queue', processor, {
embedded: true,
batchSize: 100, // Request up to 100; free concurrency slots cap the pull
pollTimeout: 5000, // Wait up to 5s for jobs (long polling)
});
```
```typescript
const worker = new Worker('queue', processor, {
embedded: false,
batchSize: 100, // Request up to 100; free concurrency slots cap the pull
pollTimeout: 5000, // Wait up to 5s for jobs (long polling)
});
```
```python
worker = Worker(
"queue",
process,
batch_size=100, # Request up to 100; free concurrency slots cap the pull
poll_timeout_ms=5000, # Wait up to 5s for jobs (long polling)
)
```
```php
$worker = new Worker('queue', $processor, [
'batchSize' => 100, // Always requests 100; jobs are leased at once and processed sequentially
'pollTimeoutMs' => 5000, // Wait up to 5s for jobs (long polling)
]);
```
```go
worker := bunqueue.NewWorker("queue", processor, bunqueue.WorkerOptions{
BatchSize: 100, // Request up to 100; free slots cap the pull
PollTimeoutMs: 5000, // Wait up to 5s for jobs (long polling)
})
```
```rust
let worker = Worker::new("queue", processor, WorkerOptions {
batch_size: 100, // Request up to 100; free slots cap the pull
poll_timeout_ms: 5_000, // Wait up to 5s for jobs (long polling)
..Default::default()
});
```
```elixir
worker =
Bunqueue.Worker.new("queue", handler,
batch_size: 100, # Request up to 100; free slots cap the pull
poll_timeout: 5_000 # Wait up to 5s for jobs (long polling)
)
```
Bulk pushes wake idle long-polling workers immediately, they never wait out the timeout.
## Native processor batches (Bun)
`batchSize` above optimizes transport but still invokes the processor once per
job. The shared TypeScript `batch` option changes the processor contract: one call
owns several independently leased jobs.
```typescript
const worker = new Worker(
'webhooks',
async (leadingJob) => {
const jobs = leadingJob.getBatch?.() ?? [leadingJob];
for (const job of jobs) {
try {
await deliver(job.data);
} catch (error) {
job.setAsFailed?.(error instanceof Error ? error : new Error(String(error)));
}
}
return { received: jobs.length };
},
{
concurrency: 4,
batch: { size: 100, minSize: 10, timeout: 250, groupAffinity: true },
}
);
```
Here `concurrency: 4` means up to four processor invocations, each with at most
100 jobs. `size` is 1..1000; `minSize` defaults to 1 and cannot exceed `size`.
When a global Worker `limiter` is present, `minSize` also cannot exceed
`limiter.max`; larger maximum batches remain valid and run in bounded chunks.
With `minSize` and no positive `timeout`, the Worker waits indefinitely for the
minimum. After a positive timeout it runs a partial batch. `groupAffinity`
keeps every member on the same job-group ID. Without affinity, any batch that
contains grouped jobs starts with the members already available instead of
waiting for `minSize`.
A Worker `limiter` still counts job starts, not processor calls: a ready batch
atomically consumes one slot per member, and a batch waiting for `minSize`
consumes no slots. If the processor throws, it is invoked once and the same
failure is applied to every member — including members already marked with
`setAsFailed()`, whose per-member error only wins when the shared invocation
resolves.
Cancellation is shared by the processor invocation. Cancelling or timing out
any active member aborts the processor context signal, so cooperative processor
code can stop the whole batch; every member then follows that shared outcome.
The processor's return value completes every member not marked with
`setAsFailed(error)`. Each member still has its own token, events, retry budget,
and final transition; one selective failure does not fail the rest of the
batch. Both TypeScript packages support processor batches; the other SDKs
retain transport batch pulling.
## Where to go next
| Guide | What it covers |
|---|---|
| [Worker](/guide/worker/) | Create a worker and process your first job |
| [The Job Object Inside a Worker Processor](/guide/worker/job-object/) | Everything the processor receives and can do |
| [Worker Events](/guide/worker/events/) | completed, failed, stalled and the rest |
| [Worker Error Handling, Retries and Backoff](/guide/worker/errors/) | Retries, backoff, timeouts and giving up |
| [Worker Lifecycle](/guide/worker/lifecycle/) | Pause, resume and shut down without losing work |
| [Heartbeats, Stall Detection and Lock Ownership](/guide/worker/stalls/) | Heartbeats, stall recovery and lock ownership |
| [SandboxedWorker](/guide/worker/sandboxed/) | Experimental isolation for CPU-heavy handlers |
| [WorkerOptions Reference](/guide/worker/options/) | Every WorkerOptions field, with defaults |