Ce post est le premier article de la série #FridayTechnical. Nous essaierons de publier un contenu technique intéressant sur les langages, les bibliothèques et les outils que nous utilisons pour construire notre produit.
Notre stack technologique est principalement basée sur Ruby et Go. À un moment donné, nous avons dû faire communiquer ces langages entre eux. Cet article se concentre sur la gestion des tâches asynchrones entre eux.
Dans toute infrastructure qui n'est pas composée d'un seul processus, des travailleurs en arrière-plan sont nécessaires. Sidekiq est un outil assez populaire pour y parvenir dans l'écosystème ruby. Il stocke son état dans une base de données redis, commune aux clients et aux travailleurs. Chez Appsdeck, notre infrastructure est plus compliquée que cela, et il est nécessaire pour nous d'envoyer des tâches asynchrones d'un service à l'autre.
Go-Workers (https://github.com/jrallison/go-workers) est un outil compatible avec Sidekiq pour gérer les tâches avec Go, utilisant presque les mêmes DSL. Nous verrons plus tard que, grâce à certaines contributions que nous avons effectuées, il fait très bien son travail.
Travailleur
Le travail du travailleur est d'écouter les tâches. Il interroge régulièrement les files d'attente que vous avez configurées et exécute les tâches qu'il trouve.
package main
import (
"fmt"
"github.com/Scalingo/go-workers"
)
func main() {
workers.Configure(map[string]string{
"process": "worker1", "password": "secret",
"server": "localhost:6379", "namespace": "goworkers",
}) workers.Process("myqueue", MyGoWorker, 10)
workers.Run()
}
func MyGoWorker(msg *workers.Msg) {
fmt.Println("running task", msg
package main
import (
"fmt"
"github.com/Scalingo/go-workers"
)
func main() {
workers.Configure(map[string]string{
"process": "worker1", "password": "secret",
"server": "localhost:6379", "namespace": "goworkers",
}) workers.Process("myqueue", MyGoWorker, 10)
workers.Run()
}
func MyGoWorker(msg *workers.Msg) {
fmt.Println("running task", msg
package main
import (
"fmt"
"github.com/Scalingo/go-workers"
)
func main() {
workers.Configure(map[string]string{
"process": "worker1", "password": "secret",
"server": "localhost:6379", "namespace": "goworkers",
}) workers.Process("myqueue", MyGoWorker, 10)
workers.Run()
}
func MyGoWorker(msg *workers.Msg) {
fmt.Println("running task", msg
package main
import (
"fmt"
"github.com/Scalingo/go-workers"
)
func main() {
workers.Configure(map[string]string{
"process": "worker1", "password": "secret",
"server": "localhost:6379", "namespace": "goworkers",
}) workers.Process("myqueue", MyGoWorker, 10)
workers.Run()
}
func MyGoWorker(msg *workers.Msg) {
fmt.Println("running task", msg
Tout d'abord, la connexion redis et l'identité du processus doivent être configurées. Ensuite, un ou plusieurs appels à la méthode Process définissent les files d'attente et les fonctions de travailleur qui leur sont associées.
La méthode du travailleur reçoit un *workers.Msg en argument. C'est en fait un wrapper d'un *simplejson.Json contenant les arguments.
require "sidekiq"
Sidekiq::configure_server do |config|
config.redis = { url: "redis://:secret@localhost:6379", namespace: "goworkers" }
end
Sidekiq.options[:queues] = ["myqueue"]
class MyRubyWorker
include Sidekiq::Worker
def perform(str)
puts "Hello from Sidekiq: #{str}"
end
endrequire "sidekiq"
Sidekiq::configure_server do |config|
config.redis = { url: "redis://:secret@localhost:6379", namespace: "goworkers" }
end
Sidekiq.options[:queues] = ["myqueue"]
class MyRubyWorker
include Sidekiq::Worker
def perform(str)
puts "Hello from Sidekiq: #{str}"
end
endrequire "sidekiq"
Sidekiq::configure_server do |config|
config.redis = { url: "redis://:secret@localhost:6379", namespace: "goworkers" }
end
Sidekiq.options[:queues] = ["myqueue"]
class MyRubyWorker
include Sidekiq::Worker
def perform(str)
puts "Hello from Sidekiq: #{str}"
end
endrequire "sidekiq"
Sidekiq::configure_server do |config|
config.redis = { url: "redis://:secret@localhost:6379", namespace: "goworkers" }
end
Sidekiq.options[:queues] = ["myqueue"]
class MyRubyWorker
include Sidekiq::Worker
def perform(str)
puts "Hello from Sidekiq: #{str}"
end
endLe processus est assez similaire avec Sidekiq, sauf que les classes Worker sont automatiquement enregistrées lorsqu'on inclut Sidekiq::Worker.
Client
Le client est le processus qui ajoute des tâches aux files d'attente. La configuration est similaire, mais au lieu d'enregistrer des travailleurs, des tâches sont mises en file d'attente dans les files.
package main
import "github.com/Scalingo/go-workers"
func main() {
workers.Configure(map[string]string{
"process": "client1", "password": "secret",
"server": "localhost:6379", "namespace": "goworkers", })
workers.Enqueue("myqueue", "MyRubyWorker", []string{"hello"
package main
import "github.com/Scalingo/go-workers"
func main() {
workers.Configure(map[string]string{
"process": "client1", "password": "secret",
"server": "localhost:6379", "namespace": "goworkers", })
workers.Enqueue("myqueue", "MyRubyWorker", []string{"hello"
package main
import "github.com/Scalingo/go-workers"
func main() {
workers.Configure(map[string]string{
"process": "client1", "password": "secret",
"server": "localhost:6379", "namespace": "goworkers", })
workers.Enqueue("myqueue", "MyRubyWorker", []string{"hello"
package main
import "github.com/Scalingo/go-workers"
func main() {
workers.Configure(map[string]string{
"process": "client1", "password": "secret",
"server": "localhost:6379", "namespace": "goworkers", })
workers.Enqueue("myqueue", "MyRubyWorker", []string{"hello"
Voici la principale différence entre le flux de travail standard de Sidekiq. Au lieu d'utiliser MyRubyWorker.enqueue "arg1", il est nécessaire d'utiliser Sidekiq::Client.push. La raison est simple, les classes ne sont pas définies dans le programme Ruby, mais dans le travailleur Go. (nous pourrions définir des classes vides, mais c'est, à mon avis, maladroit.)
require "sidekiq"
Sidekiq::configure_client do |config|
config.redis = { url: "redis://:secret@localhost:6379", namespace: "goworkers" }
end
Sidekiq::Client.push "queue" => "myqueue", "class" => "MyGoWorker", "args" => ["hello"
require "sidekiq"
Sidekiq::configure_client do |config|
config.redis = { url: "redis://:secret@localhost:6379", namespace: "goworkers" }
end
Sidekiq::Client.push "queue" => "myqueue", "class" => "MyGoWorker", "args" => ["hello"
require "sidekiq"
Sidekiq::configure_client do |config|
config.redis = { url: "redis://:secret@localhost:6379", namespace: "goworkers" }
end
Sidekiq::Client.push "queue" => "myqueue", "class" => "MyGoWorker", "args" => ["hello"
require "sidekiq"
Sidekiq::configure_client do |config|
config.redis = { url: "redis://:secret@localhost:6379", namespace: "goworkers" }
end
Sidekiq::Client.push "queue" => "myqueue", "class" => "MyGoWorker", "args" => ["hello"
Planification des tâches
Le package jrallison/go-workers fonctionne réellement mais manque de certaines fonctionnalités, en particulier, la mise en file d'attente des tâches planifiées. Par exemple, si vous souhaitez envoyer un e-mail à vos utilisateurs à 9 heures du matin, il est important d'avoir cette fonctionnalité. Notre fork (https://github.com/Scalingo/go-workers) du projet (en attente de PR pour être inclus en amont) a ces fonctionnalités.
package main
import (
"time"
"github.com/Scalingo/go-workers"
)
func main() {
workers.Configure(map[string]string{
"process": "client1", "password": "secret",
"server": "localhost:6379", "namespace": "goworkers",
})
workers.EnqueueIn("myqueue", "MyRubyWorker", 2*60, []string{"in two minutes"})
now := time.Now()
hoursTo9 := (time.Duration(9 - now.Hour())) * time.Hour
at := now.AddDate(0, 0, 1).Truncate(time.Hour).Add(hoursTo9).Unix()
workers.EnqueueAt("myqueue", "MyGoWorker", at, []string{"tomorrow at 9"
package main
import (
"time"
"github.com/Scalingo/go-workers"
)
func main() {
workers.Configure(map[string]string{
"process": "client1", "password": "secret",
"server": "localhost:6379", "namespace": "goworkers",
})
workers.EnqueueIn("myqueue", "MyRubyWorker", 2*60, []string{"in two minutes"})
now := time.Now()
hoursTo9 := (time.Duration(9 - now.Hour())) * time.Hour
at := now.AddDate(0, 0, 1).Truncate(time.Hour).Add(hoursTo9).Unix()
workers.EnqueueAt("myqueue", "MyGoWorker", at, []string{"tomorrow at 9"
package main
import (
"time"
"github.com/Scalingo/go-workers"
)
func main() {
workers.Configure(map[string]string{
"process": "client1", "password": "secret",
"server": "localhost:6379", "namespace": "goworkers",
})
workers.EnqueueIn("myqueue", "MyRubyWorker", 2*60, []string{"in two minutes"})
now := time.Now()
hoursTo9 := (time.Duration(9 - now.Hour())) * time.Hour
at := now.AddDate(0, 0, 1).Truncate(time.Hour).Add(hoursTo9).Unix()
workers.EnqueueAt("myqueue", "MyGoWorker", at, []string{"tomorrow at 9"
package main
import (
"time"
"github.com/Scalingo/go-workers"
)
func main() {
workers.Configure(map[string]string{
"process": "client1", "password": "secret",
"server": "localhost:6379", "namespace": "goworkers",
})
workers.EnqueueIn("myqueue", "MyRubyWorker", 2*60, []string{"in two minutes"})
now := time.Now()
hoursTo9 := (time.Duration(9 - now.Hour())) * time.Hour
at := now.AddDate(0, 0, 1).Truncate(time.Hour).Add(hoursTo9).Unix()
workers.EnqueueAt("myqueue", "MyGoWorker", at, []string{"tomorrow at 9"
Au lieu d'utiliser workers.Enqueue, vous pouvez maintenant utiliser workers.EnqueueIn ou workers.EnqueueAt (de la même manière que vous utiliseriez enqueue_in ou enqueue_at avec Sidekiq. L'exemple précédent utilise ces fonctions et calcule la date de demain à 9h00.
Conclusion
Avec go-workers, il est vraiment facile d'interfacer vos tâches entre Go et Ruby pour profiter du meilleur des deux mondes. Par exemple : de jolis gemmes front-end pour le rendu HTML/email pour Ruby et un traitement efficace des tâches concurrentes pour Go.
Liens
Exemples de code : https://github.com/Scalingo/go-workers-examples Bibliothèque de travailleur Go : https://github.com/Scalingo/go-workers Sidekiq : http://sidekiq.org
Crédits
Image Gopher par Renee French (Creative Commons Attributions 3.0) Logo Sidekiq © 2014 Contributed Systems LLC, contribsys.com
– Léo Unbekandt, CTO @ Appsdeck
