K60至尊版狂暴引擎2.0加持：超177萬跑分斬獲性能第一
Redmi的后性能時代戰(zhàn)略發(fā)布會今天下午如期舉辦，在本次發(fā)布會上，Redmi公布了多項關(guān)于和聯(lián)發(fā)科的深度合作，以及新機K60 Ultra在軟件和硬件方面的特性，例如：“K60 至尊版，雙芯旗艦
Asynq 是一個 Go 庫，用于對任務進行排隊并與工作人員異步處理它們。
.├── conf│ └── redis.conf└── docker-compose.ymlversion: '3.8'services: myredis: container_name: myredis image: redis:6.2.5 #6.0.6 restart: always ports: - 6379:6379 privileged: true environment: # 時區(qū)上海 TZ: Asia/Shanghai command: redis-server /etc/redis/redis.conf --appendonly yes volumes: - $PWD/data:/data - $PWD/conf/redis.conf:/etc/redis/redis.conf # networks: # - myweb#networks: # myweb: # driver: bridge#開啟保護protected-mode yes#開啟遠程連接 #bind 127.0.0.1 #自定義密碼requirepass 123456 port 6379timeout 0# 900s內(nèi)至少一次寫操作則執(zhí)行bgsave進行RDB持久化save 900 1 save 300 10save 60 10000# rdbcompression ；默認值是yes。對于存儲到磁盤中的快照，可以設(shè)置是否進行壓縮存儲。如果是的話，redis會采用LZF算法進行壓縮。如果你不想消耗CPU來進行壓縮的話，可以設(shè)置為關(guān)閉此功能，但是存儲在磁盤上的快照會比較大。rdbcompression yes# dbfilename ：設(shè)置快照的文件名，默認是 dump.rdbdbfilename dump.rdb# dir：設(shè)置快照文件的存放路徑，這個配置項一定是個目錄，而不能是文件名。使用上面的 dbfilename 作為保存的文件名。dir /data# 默認redis使用的是rdb方式持久化，這種方式在許多應用中已經(jīng)足夠用了。但是redis如果中途宕機，會導致可能有幾分鐘的數(shù)據(jù)丟失，根據(jù)save來策略進行持久化，Append Only File是另一種持久化方式， 可以提供更好的持久化特性。Redis會把每次寫入的數(shù)據(jù)在接收后都寫入appendonly.aof文件，每次啟動時Redis都會先把這個文件的數(shù)據(jù)讀入內(nèi)存里，先忽略RDB文件。默認值為no。appendonly yes# appendfilename ：aof文件名，默認是"appendonly.aof"# appendfsync：aof持久化策略的配置；no表示不執(zhí)行fsync，由操作系統(tǒng)保證數(shù)據(jù)同步到磁盤，速度最快；always表示每次寫入都執(zhí)行fsync，以保證數(shù)據(jù)同步到磁盤；everysec表示每秒執(zhí)行一次fsync，可能會導致丟失這1s數(shù)據(jù)appendfsync everysec.|-- README.md|-- cmd| `-- main.go # 啟動消息者監(jiān)聽|-- go.mod|-- go.sum|-- test.go # 生成者 發(fā)送測試數(shù)據(jù)`-- test_delivery |-- client # 具體發(fā)送測試數(shù)據(jù)的邏輯 | `-- client.go `-- test_delivery.go # 消息者，執(zhí)行任務具體處理邏輯Asynq 使用 Redis 作為消息代理。client.go 和 main.go 都需要連接到 Redis 進行寫入和讀取。我們將使用 asynq.RedisClientOpt 指定如何連接到本地 Redis 實例。
asynq.RedisClientOpt{ Addr: "127.0.0.1:6379", Password: "", DB: 2,}type Task struct { // 一個簡單的字符串值，表示要執(zhí)行的任務的類型. typename string // 有效載荷保存執(zhí)行任務所需的數(shù)據(jù)，有效負載值必須是可序列化的. payload []byte // 保存任務的選項. opts []Option // 任務的結(jié)果編寫器. w *ResultWriter}package test_deliveryimport ( "context" "encoding/json" "fmt" "github.com/hibiken/asynq" "log")const ( TypeEmailDelivery = "email:deliver")// EmailDeliveryPayload 異步任務需要傳遞的數(shù)據(jù)結(jié)構(gòu)type EmailDeliveryPayload struct { UserID int TemplateID string DataStr string}// NewEmailDeliveryTask 異步任務需要傳遞的數(shù)據(jù)func NewEmailDeliveryTask(userID int, tmplID, dataStr string) (*asynq.Task, error) { payload, err := json.Marshal(EmailDeliveryPayload{UserID: userID, TemplateID: tmplID, DataStr: dataStr}) if err != nil { fmt.Println(err) return nil, err } return asynq.NewTask(TypeEmailDelivery, payload), nil}// HandleEmailDeliveryTask 發(fā)送email處理邏輯func HandleEmailDeliveryTask(ctx context.Context, t *asynq.Task) error { //接收任務數(shù)據(jù) var p EmailDeliveryPayload if err := json.Unmarshal(t.Payload(), &p); err != nil { return fmt.Errorf("json.Unmarshal failed: %v: %w", err, asynq.SkipRetry) } //邏輯處理start... log.Printf("Sending Email to User: user_id=%d, template_id=%s data_str:%s", p.UserID, p.TemplateID, p.DataStr) return nil}在應用程序代碼中，導入上述包并用于Client將任務放入隊列中。
package clientimport ( "asynq_task/test_delivery" "fmt" "github.com/hibiken/asynq" "log" "time")func EmailDeliveryTaskAdd(i int) { client := asynq.NewClient(asynq.RedisClientOpt{ Addr: "192.168.0.120:6379", Password: "123456", DB: 2, }) defer client.Close() // 初使貨需要傳遞的數(shù)據(jù) task, err := test_delivery.NewEmailDeliveryTask(42, fmt.Sprintf("some:template:id:%d", i), `{"name":"lisi"}`) if err != nil { log.Fatalf("could not create task: %v", err) } // 任務入隊 //info, err := client.Enqueue(task) //info, err := client.Enqueue(task, time.Now()) // 延遲執(zhí)行 info, err := client.Enqueue(task, asynq.ProcessIn(3*time.Second)) // MaxRetry 重度次數(shù) Timeout超時時間 //info, err = client.Enqueue(task, asynq.MaxRetry(10), asynq.Timeout(3*time.Second)) if err != nil { log.Fatalf("could not enqueue task: %v", err) } log.Printf("enqueued task: id=%s queue=%s", info.ID, info.Queue)}接下來，啟動一個工作服務器以在后臺處理這些任務。要啟動后臺工作人員，使用Server并提供您Handler來處理任務。可以選擇使用ServeMux來創(chuàng)建處理程序，就像使用net/httpHandler 一樣。
package mainimport ( "asynq_task/test_delivery" "github.com/hibiken/asynq" "log")func main() { srv := asynq.NewServer( asynq.RedisClientOpt{ Addr: "192.168.0.120:6379", Password: "123456", DB: 2, }, asynq.Config{ // 每個進程并發(fā)執(zhí)行的worker數(shù)量 Concurrency: 5, // Optionally specify multiple queues with different priority. Queues: map[string]int{ "critical": 6, "default": 3, "low": 1, }, // See the godoc for other configuration options }, ) mux := asynq.NewServeMux() mux.HandleFunc(test_delivery.TypeEmailDelivery, test_delivery.HandleEmailDeliveryTask) if err := srv.Run(mux); err != nil { log.Fatalf("could not run server: %v", err) }}package mainimport ( "asynq_task/test_delivery/client" "time")func main() { for i := 0; i < 3; i++ { client.EmailDeliveryTaskAdd(i) time.Sleep(time.Second * 3) }}結(jié)果如下：
消息者 go run main.go
$ go run main.go asynq: pid=12092 2023/02/02 23:18:04.161872 INFO: Starting processingasynq: pid=12092 2023/02/02 23:18:04.161872 INFO: Send signal TERM or INT to terminate the process2023/02/03 07:18:14 Sending Email to User: user_id=42, template_id=some:template:id:0 data_str:{"name":"lisi"}2023/02/03 07:18:19 Sending Email to User: user_id=42, template_id=some:template:id:1 data_str:{"name":"lisi"}2023/02/03 07:18:19 Sending Email to User: user_id=42, template_id=some:template:id:2 data_str:{"name":"lisi"}生產(chǎn)者 go run test.go
$ go run test.go 2023/02/03 07:18:09 enqueued task: id=5d998c6b-3978-4a25-a096-6e564e032359 queue=default2023/02/03 07:18:12 enqueued task: id=74a5fea4-d4d4-465f-b310-31981e472f6a queue=default2023/02/03 07:18:15 enqueued task: id=41c46b7b-ea78-4abc-878a-ea65e3859e28 queue=default如果異步服務突然被暫停，正在執(zhí)行的異步任務會push到隊列中，下次啟動的時候自動執(zhí)行。
我們可以將一個異步任務中途sleep幾秒，發(fā)送一個異步任務，任務沒執(zhí)行完中途停掉任務測試出結(jié)果：
圖片
再次啟動異步任務服務，發(fā)現(xiàn)這個任務被重新執(zhí)行。
client.Enqueue(t1, time.Now())
client.Enqueue(t2, asynq.ProcessIn(time.Now().Add(2 * time.Hour)))
client.Enqueue(task, asynq.MaxRetry(5))
4-1：使用TaskID選項：自行生成唯一的任務 ID
_, err := client.Enqueue(task, asynq.TaskID("mytaskid"))// Second task will fail, err is ErrTaskIDConflict (assuming that the first task didn't get processed yet)_, err = client.Enqueue(task, asynq.TaskID("mytaskid"))4-2：使用Unique選項：讓 Asynq 為任務創(chuàng)建唯一性鎖
err := c.Enqueue(t1, asynq.Unique(time.Hour))另外，asynq 異步任務提供了命令行工具和 Asynqmon 用于監(jiān)控和管理 Asynq 異步任務和隊列。WebUI 可以通過傳遞兩個標志來啟用與 Prometheus 的集成。
#asynqmon asynq延遲隊列、定時隊列的webui asynqmon: image: hibiken/asynqmon:latest container_name: asynqmon ports: - 8980:8080 command: - '--redis-addr=192.168.0.120:6379' - '--redis-password=123456' - '--redis-db=2' restart: always# networks:# - looklook_net# depends_on:# - redis啟動服務：docker-compose up
訪問：http://192.168.0.120:8980/
圖片
原文鏈接：http://haimait.top/docs/golang/asynq
本文轉(zhuǎn)載自微信公眾號「 程序員升級打怪之旅」，作者「 海馬&王中陽Go」，可以通過以下二維碼關(guān)注。
轉(zhuǎn)載本文請聯(lián)系「 程序員升級打怪之旅」公眾號。
本文鏈接：http://www.www897cc.com/showinfo-26-16838-0.htmlGo異步任務解決方案：Asynq
聲明：本網(wǎng)頁內(nèi)容旨在傳播知識，若有侵權(quán)等問題請及時與本網(wǎng)聯(lián)系，我們將在第一時間刪除處理。郵件：2376512515@qq.com
Copyright ? 2016-2023 天津谷騏科技有限公司 版權(quán)所有 sitemap.xml
違法及侵權(quán)請聯(lián)系：2376512515@qq.com 津ICP備18001702號
津公網(wǎng)安備 12010102000574號