fix: standardize variable naming and adjust loader sleep duration for consistency
This commit is contained in:
@@ -9,12 +9,14 @@ import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
const (
|
const (
|
||||||
NumExtractors int = 2
|
numExtractors int = 2
|
||||||
NumLoaders int = 4
|
numTransformers int = numExtractors
|
||||||
ChunkSize int = 20
|
numLoaders int = 4
|
||||||
totalRecords int = 500
|
chunkSize int = 20
|
||||||
queueSize int = 10
|
totalRecords int = 1000
|
||||||
recordsPerExtractor int = totalRecords / NumExtractors
|
extractorsQueueSize int = 10
|
||||||
|
transformersQueueSize int = 10
|
||||||
|
recordsPerExtractor int = totalRecords / numExtractors
|
||||||
)
|
)
|
||||||
|
|
||||||
type Record struct {
|
type Record struct {
|
||||||
@@ -89,18 +91,18 @@ func Transformer(id int, in <-chan []Record, out chan<- []Record) {
|
|||||||
func Loader(id int, in <-chan []Record) {
|
func Loader(id int, in <-chan []Record) {
|
||||||
for chunk := range in {
|
for chunk := range in {
|
||||||
fmt.Printf("[Loader %d] Procesando lote de %d registros...\n", id, len(chunk))
|
fmt.Printf("[Loader %d] Procesando lote de %d registros...\n", id, len(chunk))
|
||||||
time.Sleep(randomDurationMs(100, 3000))
|
time.Sleep(randomDurationMs(100, 2000))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func main() {
|
func main() {
|
||||||
chChunksExtract := make(chan []Record, queueSize)
|
chChunksExtract := make(chan []Record, extractorsQueueSize)
|
||||||
chChunksTransform := make(chan []Record, queueSize)
|
chChunksTransform := make(chan []Record, transformersQueueSize)
|
||||||
|
|
||||||
var wgExtractors sync.WaitGroup
|
var wgExtractors sync.WaitGroup
|
||||||
for i := 1; i <= NumExtractors; i++ {
|
for i := 1; i <= numExtractors; i++ {
|
||||||
wgExtractors.Go(func() {
|
wgExtractors.Go(func() {
|
||||||
Extractor(i, ChunkSize, chChunksExtract)
|
Extractor(i, chunkSize, chChunksExtract)
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -111,7 +113,7 @@ func main() {
|
|||||||
}()
|
}()
|
||||||
|
|
||||||
var wgTransformers sync.WaitGroup
|
var wgTransformers sync.WaitGroup
|
||||||
for i := 1; i <= NumExtractors; i++ {
|
for i := 1; i <= numTransformers; i++ {
|
||||||
wgTransformers.Go(func() {
|
wgTransformers.Go(func() {
|
||||||
Transformer(i, chChunksExtract, chChunksTransform)
|
Transformer(i, chChunksExtract, chChunksTransform)
|
||||||
})
|
})
|
||||||
@@ -124,7 +126,7 @@ func main() {
|
|||||||
}()
|
}()
|
||||||
|
|
||||||
var wgLoaders sync.WaitGroup
|
var wgLoaders sync.WaitGroup
|
||||||
for i := 1; i <= NumLoaders; i++ {
|
for i := 1; i <= numLoaders; i++ {
|
||||||
wgLoaders.Go(func() {
|
wgLoaders.Go(func() {
|
||||||
Loader(i, chChunksTransform)
|
Loader(i, chChunksTransform)
|
||||||
})
|
})
|
||||||
|
|||||||
Reference in New Issue
Block a user