Redisをタスクキューとして利用する(Golang)

        での 前へ 分散キュー的なものを自分のマシンでピュアゴーでシミュレートしてみた。Redisキューの部分をここに追加。
Redisキューには、解決しなければならない3つの問題があります。
 1. 待ち行列の構築
     RedisのRPUSH/LPOPを使用します。
 2. パラメータの受け渡し/解析 
 クライアントはJOSNパラメータをRedisに格納し、サーバーはそれを取り出して解析して返します。
 3. コネクションプーリング

      redigoがRedisコネクションプールをサポート

以下のコードは、その解決策を具体的に実装したものです。

//Redis does background task queues
//author: Xiong Chuan Liang
//date: 2015-3-25

package main

import (
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	"github.com/garyburd/redigo/redis"
)

func main() {

	r, err := newRedisPool("", "")
	if err ! = nil {
		fmt.Println(err)
		return
	}

	//put the job into the queue
	r.Enqueue()

	//Take out two jobs in turn
	r.GetJob()
	r.GetJob()
GetJob()

type RedisPool struct {
	pool *redis.
Pool }

func newRedisPool(server, password string) (*RedisPool, error) {

	if server == "" {
		server = ":6379"
	}

	pool := &redis.Pool{
		MaxIdle: 3,
		IdleTimeout: 240 * time.Second,
		Dial: func() (redis.Conn, error) {
			c, err := redis.Dial("tcp", server)
			if err ! = nil {
				return nil, err
			}

			if password ! = "" {
				if _, err := c.Do("AUTH", password); err ! = nil {
					c.Close()
					return nil, err
				}
			}
			return c, err
		},
		TestOnBorrow: func(c redis.Conn, t time.Time) error {
			_, err := c.Do("PING")
			return err
		},
	}

	return &RedisPool{pool}, nil
}

type Job struct {
	Class string `json:"Class"`
	Args []interface{} `json:"Args"`
}

//Simulate the client
func (r *RedisPool) Enqueue() error {

	c := r.pool.Get()
	defer c.Close()

	j := &Job{}
	j.Class = "mail"
	j.Args = append(j.Args, "[email protected]", "", "body", 2, true)

	j2 := &Job{}
	j2.Class = "Log"
	j2.Args = append(j2.Args, "ccc.log", "ddd.log", []int{222, 333})

	for _, v := range []*Job{j, j2} {
		b, err := json.Marshal(v)
		if err ! = nil {
			return err
		}

		_, err = c.Do("rpush", "queue", b)
		if err ! = nil {
			return err
		}
	}

	fmt.Println("[Enqueue()] succeed!")

	return nil
}

// Simulate Job Server
func (r *RedisPool) GetJob() error {
	count, err := r.QueuedJobCount()
	if err ! = nil || count == 0 {
		return errors.New("No Job. ")
	}
	fmt.Println("[GetJob()] Jobs count:", count)

	c := r.pool.Get()
	defer c.Close()

	for i := 0; i < int(count); i++ {
		reply, err := c.Do("LPOP", "queue")
		if err ! = nil {
			return err
		}

		var j Job
		decoder := json.NewDecoder(bytes.NewReader(reply.([]byte)))
		If err := decoder.Decode(&j); err ! = nil {
			return err
		}

		fmt.Println("[GetJob()] ", j.Class, " : ", j.Args)
	}
	return nil
}

func (r *RedisPool) QueuedJobCount() (int, error) {
	c := r.pool.Get()
	defer c.Close()

	lenqueue, err := c.Do("len", "queue")
	if err ! = nil {
		return 0, err
	}

	count, ok := lenqueue.(int64)
	if !ok {
		return 0, errors.New("Type conversion error! ")
	}
	return int(count), nil
}

/*
Run the result:

[Enqueue()] succeed!
[GetJob()] Jobs count: 2
[GetJob()] mail : [[email protected] body 2 true]
[GetJob()] log : [ccc.log ddd.log [222 333]]
[root@xclos src]#

*/

    Goがパラメータを取得できていることがわかります。とても基本的なものばかりです。

EMAIL：[email protected]

BLOG：http://blog.csdn.net