golang Go和MongoDB的JSON-API：最终部分

Posted 2021-05-24

package main

import (
	"encoding/json"
	"log"
	"net/http"
	"reflect"
	"time"

	"github.com/gorilla/context"
	"github.com/julienschmidt/httprouter"
	"github.com/justinas/alice"
	"gopkg.in/mgo.v2"
	"gopkg.in/mgo.v2/bson"
)

// Repo

type Tea struct {
	Id       bson.ObjectId `json:"id,omitempty" bson:"_id,omitempty"`
	Name     string        `json:"name"`
	Category string        `json:"category"`
}

type TeasCollection struct {
	Data []Tea `json:"data"`
}

type TeaResource struct {
	Data Tea `json:"data"`
}

type TeaRepo struct {
	coll *mgo.Collection
}

func (r *TeaRepo) All() (TeasCollection, error) {
	result := TeasCollection{[]Tea{}}
	err := r.coll.Find(nil).All(&result.Data)
	if err != nil {
		return result, err
	}

	return result, nil
}

func (r *TeaRepo) Find(id string) (TeaResource, error) {
	result := TeaResource{}
	err := r.coll.FindId(bson.ObjectIdHex(id)).One(&result.Data)
	if err != nil {
		return result, err
	}

	return result, nil
}

func (r *TeaRepo) Create(tea *Tea) error {
	id := bson.NewObjectId()
	_, err := r.coll.UpsertId(id, tea)
	if err != nil {
		return err
	}

	tea.Id = id

	return nil
}

func (r *TeaRepo) Update(tea *Tea) error {
	err := r.coll.UpdateId(tea.Id, tea)
	if err != nil {
		return err
	}

	return nil
}

func (r *TeaRepo) Delete(id string) error {
	err := r.coll.RemoveId(bson.ObjectIdHex(id))
	if err != nil {
		return err
	}

	return nil
}

// Errors

type Errors struct {
	Errors []*Error `json:"errors"`
}

type Error struct {
	Id     string `json:"id"`
	Status int    `json:"status"`
	Title  string `json:"title"`
	Detail string `json:"detail"`
}

func WriteError(w http.ResponseWriter, err *Error) {
	w.Header().Set("Content-Type", "application/vnd.api+json")
	w.WriteHeader(err.Status)
	json.NewEncoder(w).Encode(Errors{[]*Error{err}})
}

var (
	ErrBadRequest           = &Error{"bad_request", 400, "Bad request", "Request body is not well-formed. It must be JSON."}
	ErrNotAcceptable        = &Error{"not_acceptable", 406, "Not Acceptable", "Accept header must be set to 'application/vnd.api+json'."}
	ErrUnsupportedMediaType = &Error{"unsupported_media_type", 415, "Unsupported Media Type", "Content-Type header must be set to: 'application/vnd.api+json'."}
	ErrInternalServer       = &Error{"internal_server_error", 500, "Internal Server Error", "Something went wrong."}
)

// Middlewares

func recoverHandler(next http.Handler) http.Handler {
	fn := func(w http.ResponseWriter, r *http.Request) {
		defer func() {
			if err := recover(); err != nil {
				log.Printf("panic: %+v", err)
				WriteError(w, ErrInternalServer)
			}
		}()

		next.ServeHTTP(w, r)
	}

	return http.HandlerFunc(fn)
}

func loggingHandler(next http.Handler) http.Handler {
	fn := func(w http.ResponseWriter, r *http.Request) {
		t1 := time.Now()
		next.ServeHTTP(w, r)
		t2 := time.Now()
		log.Printf("[%s] %q %v\n", r.Method, r.URL.String(), t2.Sub(t1))
	}

	return http.HandlerFunc(fn)
}

func acceptHandler(next http.Handler) http.Handler {
	fn := func(w http.ResponseWriter, r *http.Request) {
		if r.Header.Get("Accept") != "application/vnd.api+json" {
			WriteError(w, ErrNotAcceptable)
			return
		}

		next.ServeHTTP(w, r)
	}

	return http.HandlerFunc(fn)
}

func contentTypeHandler(next http.Handler) http.Handler {
	fn := func(w http.ResponseWriter, r *http.Request) {
		if r.Header.Get("Content-Type") != "application/vnd.api+json" {
			WriteError(w, ErrUnsupportedMediaType)
			return
		}

		next.ServeHTTP(w, r)
	}

	return http.HandlerFunc(fn)
}

func bodyHandler(v interface{}) func(http.Handler) http.Handler {
	t := reflect.TypeOf(v)

	m := func(next http.Handler) http.Handler {
		fn := func(w http.ResponseWriter, r *http.Request) {
			val := reflect.New(t).Interface()
			err := json.NewDecoder(r.Body).Decode(val)

			if err != nil {
				WriteError(w, ErrBadRequest)
				return
			}

			if next != nil {
				context.Set(r, "body", val)
				next.ServeHTTP(w, r)
			}
		}

		return http.HandlerFunc(fn)
	}

	return m
}

// Main handlers

type appContext struct {
	db *mgo.Database
}

func (c *appContext) teasHandler(w http.ResponseWriter, r *http.Request) {
	repo := TeaRepo{c.db.C("teas")}
	teas, err := repo.All()
	if err != nil {
		panic(err)
	}

	w.Header().Set("Content-Type", "application/vnd.api+json")
	json.NewEncoder(w).Encode(teas)
}

func (c *appContext) teaHandler(w http.ResponseWriter, r *http.Request) {
	params := context.Get(r, "params").(httprouter.Params)
	repo := TeaRepo{c.db.C("teas")}
	tea, err := repo.Find(params.ByName("id"))
	if err != nil {
		panic(err)
	}

	w.Header().Set("Content-Type", "application/vnd.api+json")
	json.NewEncoder(w).Encode(tea)
}

func (c *appContext) createTeaHandler(w http.ResponseWriter, r *http.Request) {
	body := context.Get(r, "body").(*TeaResource)
	repo := TeaRepo{c.db.C("teas")}
	err := repo.Create(&body.Data)
	if err != nil {
		panic(err)
	}

	w.Header().Set("Content-Type", "application/vnd.api+json")
	w.WriteHeader(201)
	json.NewEncoder(w).Encode(body)
}

func (c *appContext) updateTeaHandler(w http.ResponseWriter, r *http.Request) {
	params := context.Get(r, "params").(httprouter.Params)
	body := context.Get(r, "body").(*TeaResource)
	body.Data.Id = bson.ObjectIdHex(params.ByName("id"))
	repo := TeaRepo{c.db.C("teas")}
	err := repo.Update(&body.Data)
	if err != nil {
		panic(err)
	}

	w.WriteHeader(204)
	w.Write([]byte("\n"))
}

func (c *appContext) deleteTeaHandler(w http.ResponseWriter, r *http.Request) {
	params := context.Get(r, "params").(httprouter.Params)
	repo := TeaRepo{c.db.C("teas")}
	err := repo.Delete(params.ByName("id"))
	if err != nil {
		panic(err)
	}

	w.WriteHeader(204)
	w.Write([]byte("\n"))
}

// Router

type router struct {
	*httprouter.Router
}

func (r *router) Get(path string, handler http.Handler) {
	r.GET(path, wrapHandler(handler))
}

func (r *router) Post(path string, handler http.Handler) {
	r.POST(path, wrapHandler(handler))
}

func (r *router) Put(path string, handler http.Handler) {
	r.PUT(path, wrapHandler(handler))
}

func (r *router) Delete(path string, handler http.Handler) {
	r.DELETE(path, wrapHandler(handler))
}

func NewRouter() *router {
	return &router{httprouter.New()}
}

func wrapHandler(h http.Handler) httprouter.Handle {
	return func(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
		context.Set(r, "params", ps)
		h.ServeHTTP(w, r)
	}
}

func main() {
	session, err := mgo.Dial("localhost")
	if err != nil {
		panic(err)
	}
	defer session.Close()
	session.SetMode(mgo.Monotonic, true)

	appC := appContext{session.DB("test")}
	commonHandlers := alice.New(context.ClearHandler, loggingHandler, recoverHandler, acceptHandler)
	router := NewRouter()
	router.Get("/teas/:id", commonHandlers.ThenFunc(appC.teaHandler))
	router.Put("/teas/:id", commonHandlers.Append(contentTypeHandler, bodyHandler(TeaResource{})).ThenFunc(appC.updateTeaHandler))
	router.Delete("/teas/:id", commonHandlers.ThenFunc(appC.deleteTeaHandler))
	router.Get("/teas", commonHandlers.ThenFunc(appC.teasHandler))
	router.Post("/teas", commonHandlers.Append(contentTypeHandler, bodyHandler(TeaResource{})).ThenFunc(appC.createTeaHandler))
	http.ListenAndServe(":8080", router)
}

带有 mgo 的 Go (golang) 中的 MongoDB：如何更新记录、确定更新是不是成功并在单个原子操作中获取数据？

【中文标题】带有 mgo 的 Go (golang) 中的 MongoDB：如何更新记录、确定更新是不是成功并在单个原子操作中获取数据？

【英文标题】：MongoDB in Go (golang) with mgo: How do I update a record, find out if update was successful and get the data in a single atomic operation?带有 mgo 的 Go (golang) 中的 MongoDB：如何更新记录、确定更新是否成功并在单个原子操作中获取数据？

【发布时间】：2012-07-10 04:38:26

【问题描述】：

我在 Go 下使用 MongoDB 的 mgo 驱动程序。

我的应用程序请求一个任务（仅在 Mongo 中从名为“jobs”的集合中选择一个记录），然后将自己注册为受让人以完成该任务（对同一“job”记录的更新，将自己设置为受让人）。

该程序将在多台机器上运行，都与同一个 Mongo 通信。当我的程序列出可用任务然后选择一个时，其他实例可能已经获得了该分配，并且当前分配将失败。

如何确定我读取并更新的记录在更新时是否具有特定值（在本例中为受让人）？

我正在尝试获得一项任务，无论是哪一项，所以我认为我应该先选择一个待处理的任务并尝试分配它，以防更新成功。

所以，我的查询应该是这样的：

“从集合 'jobs' 的所有记录中，更新 仅一个 具有 assignee=null 的记录，将我的 ID 设置为受让人。然后，给我该记录 所以我可以胜任这份工作。”

我如何用 Go 的 mgo 驱动程序来表达这一点？

【参考方案1】：

MongoDB 的家伙在官方文档中描述了类似的场景：http://www.mongodb.org/display/DOCS/Atomic+Operations

基本上，您所要做的就是使用assignee=null 获取任何工作。假设你得到了_id=42 的工作。然后，您可以继续在本地修改文档，方法是设置 assignee="worker1.example.com" 并使用选择器 _id=42, assignee=null 和您更新的文档调用 Collection.Update()。如果数据库仍然能够找到与此选择器匹配的文档，它将自动替换该文档。否则你会得到一个 ErrNotFound，表明另一个线程已经认领了这个任务。如果是这种情况，请再试一次。

【讨论】：

执行选择然后更新不是原子的；它执行两次往返！ findAndModify 命令将自动执行此操作。此命令的 Mongo 文档在这里：mongodb.org/display/DOCS/findAndModify+Command 在 Go 中执行此操作的 Mgo 文档在这里：go.pkgdoc.org/labix.org/v2/mgo#Query.Apply

对于这个特定的算法，选择+更新不需要是原子的。只要更新本身是原子的（基本上是 CompareAndSwap / CompareExchange 操作），一切都很好。

如果两台机器在第一次查询时选择相同的作业，其中一台将在更新时失败。当然，最终结果可能总是数据库中的数据永远不会变得不一致，但是使用两个单独操作的失败情况会导致很多不必要的来回。这是... findAndModify 命令存在的全部原因。

【参考方案2】：

我希望您在所选答案中看到了 cmets，但这种方法是不正确的。进行选择然后更新将导致往返和两台机器并在其中一台更新assignee之前获取相同的作业。您需要改用findAndModify 方法：http://www.mongodb.org/display/DOCS/findAndModify+Command

【讨论】：

实际上，我需要的解决方案是提供的 tux21b。我需要检索所有“选项”，然后选择一个，然后尝试将其分配给自己。如果不成功，我会尝试另一个。

为什么需要选择所有选项？你是说你只需要选择那个没有被占用的（又名assignee == null）？

你是对的。事实证明，我的需求与我写问题时的想法不同，但您的回答更符合问题。

mgo 通过Query.Apply 方法方便地支持 findAndModify 命令。详情请看下面的答案。

【参考方案3】：

这是一个老问题，但以防万一有人还在家里看，Query.Apply 方法很好地支持了这一点。它确实运行 findAndModify 命令，如另一个答案中所示，但它方便地隐藏在 Go goodness 后面。

文档中的示例与此处的问题几乎完全匹配：

change := mgo.Change
        Update: bson.M"$inc": bson.M"n": 1,
        ReturnNew: true,

info, err = col.Find(M"_id": id).Apply(change, &doc)
fmt.Println(doc.N)