命令记不住

登陆

Golang 接入 nsq

  |   129 浏览

Golang 接入 nsq

1.消息设计

场景分析

topic name/action 触发场景 消费者行为
article_modified 博客文章的新增/修改 更新ES对应记录
aricle_removed 博客文章删除 删除ES对应记录

消息设计必须有文档！

实现步骤

  1. 定义消息格式
  2. 在更新数据库的代码段中，加入发送消息的动作
  3. 开发一个独立的服务（应用），消费消息，更新ES

2.nsq工具包

安装golang package
go get github.com/nsqio/go-nsq
nsqutil
package nsqutil

import (
	"encoding/json"
	"fmt"

	"github.com/nsqio/go-nsq"
)

var producer *nsq.Producer

// 初始化生产者
func InitProducer(str string) {
	var err error
	fmt.Println("address: ", str)
	producer, err = nsq.NewProducer(str, nsq.NewConfig())
	if err != nil {
		panic(err)
	}
}

// 发布String消息
func PublishString(topic string, message string) error {
	var err error
	if producer != nil {
		if message == "" { //不能发布空串，否则会导致error
			return nil
		}
		err = producer.Publish(topic, []byte(message)) // 发布消息
		return err
	}
	return fmt.Errorf("producer is nil")
}

// 发布Json消息
func PublishJson(topic string, qMsg interface{}) error {
	message, err := json.Marshal(qMsg)
	if err != nil {
		return err
	}
	fmt.Printf("send: %s\n", string(message))

	if producer != nil {
		if len(message) == 0 { //不能发布空串，否则会导致error
			return nil
		}
		err = producer.Publish(topic, message) // 发布消息
		return err
	}
	return fmt.Errorf("producer is nil")
}

func init() {
	InitProducer("127.0.0.1:4150")
}
消息体
package nsqutil

import "fmt"

const (
	TopicEs                     = "es"
	ActionArticleModified int32 = 1 // 文章创建修改
	ActionArticleRemoved  int32 = 3 // 文章删除

)

type MessageEs struct {
	Action int32 `json:"action"`
	Id     int64 `json:"id"`
}

func SendEsMessage(action int32, id int64) error {
	qMsg := &MessageEs{
		Action: action,
		Id:     id,
	}
	if err := PublishJson(TopicEs, qMsg); err != nil {
		fmt.Errorf("send error: %s", err.Error())
		return err
	}
	return nil
}

3.添加消息发送代码

grpcdemo\server\blogdataimpl.go

package main

import (
	"fmt"
	"models"
	"nsqutil"
	pb "protoes"
	"time"

	"context"
)

// 对象要和proto内定义的服务一致
type BlogDataServer struct {
}

// UserInsert
func (s *BlogDataServer) UserInsert(ctx context.Context, in *pb.UserInsertRequest) (*pb.UserInsertResponse, error) {
	fmt.Printf("UserInsert: %s\n", time.Now().Format(TIME_FORMAT))
	return &pb.UserInsertResponse{}, nil
}

// UserQueryWithUsername
func (s *BlogDataServer) UserQueryWithUsername(ctx context.Context, in *pb.UserQueryWithUsernameRequest) (*pb.UserQueryWithUsernameResponse, error) {
	now := time.Now()
	fmt.Printf("UserQueryWithUsername: %s\n", now.Format(TIME_FORMAT))

	id := models.QueryUserWithUsername(in.Username)
	return &pb.UserQueryWithUsernameResponse{
		Id: int64(id),
	}, nil
}

// UserQueryWithParam
func (s *BlogDataServer) UserQueryWithParam(ctx context.Context, in *pb.UserQueryWithParamRequest) (*pb.UserQueryWithParamResponse, error) {
	now := time.Now()
	fmt.Printf("UserQueryWithParam: %s\n", now.Format(TIME_FORMAT))
	id := models.QueryUserWithParam(in.Username, in.Password)
	fmt.Printf("duration: %v\n", time.Now().Sub(now))
	return &pb.UserQueryWithParamResponse{
		Id: int64(id),
	}, nil
}

// NSQ
// ArticleAdd
func (s *BlogDataServer) ArticleAdd(ctx context.Context, in *pb.ArticleAddRequest) (*pb.ArticleAddResponse, error) {
	now := time.Now()
	fmt.Printf("ArticleAdd: %s\n", now.Format(TIME_FORMAT))
	id, count, err := models.AddArticle(models.ArticleFromPb(in.Article))
	msg := ""
	if err != nil {
		msg = err.Error()
	} else {
		nsqutil.SendEsMessage(nsqutil.ActionArticleModified, id)
	}
	fmt.Printf("duration: %v\n", time.Now().Sub(now))
	return &pb.ArticleAddResponse{
		Id:       id,
		TotalNum: int64(count),
		Msg:      msg,
	}, nil
}

// ArticleQueryWithCon
func (s *BlogDataServer) ArticleQueryWithCon(ctx context.Context, in *pb.ArticleQueryWithConRequest) (*pb.ArticleQueryWithConResponse, error) {
	now := time.Now()
	fmt.Printf("ArticleQueryWithCon: %s\n", now.Format(TIME_FORMAT))
	list, err := models.QueryArticlesWithCon(in.Sql)
	msg := ""
	if err != nil {
		msg = err.Error()
	}
	fmt.Printf("duration: %v\n", time.Now().Sub(now))
	return &pb.ArticleQueryWithConResponse{
		Article: models.ArticleListToPb(list),
		Msg:     msg,
	}, nil
}

// ArticleQueryRowNum
func (s *BlogDataServer) ArticleQueryRowNum(ctx context.Context, in *pb.ArticleQueryRowNumRequest) (*pb.ArticleQueryRowNumResponse, error) {
	now := time.Now()
	fmt.Printf("ArticleQueryRowNum: %s\n", now.Format(TIME_FORMAT))
	fmt.Printf("duration: %v\n", time.Now().Sub(now))
	return &pb.ArticleQueryRowNumResponse{
		Count: int64(models.QueryArticleRowNum()),
	}, nil
}

// ArticleQueryWithId
func (s *BlogDataServer) ArticleQueryWithId(ctx context.Context, in *pb.ArticleQueryWithIdRequest) (*pb.ArticleQueryWithIdResponse, error) {
	now := time.Now()
	fmt.Printf("ArticleQueryWithId: %s\n", now.Format(TIME_FORMAT))
	article := models.QueryArticleWithId(int(in.Id))
	fmt.Printf("duration: %v\n", time.Now().Sub(now))
	return &pb.ArticleQueryWithIdResponse{
		Article: models.ArticleToPb(article),
	}, nil
}

// ArticleUpdate
func (s *BlogDataServer) ArticleUpdate(ctx context.Context, in *pb.ArticleUpdateRequest) (*pb.ArticleUpdateResponse, error) {
	now := time.Now()
	fmt.Printf("ArticleUpdate: %s\n", now.Format(TIME_FORMAT))
	id := in.Article.Id
	count, err := models.UpdateArticle(models.ArticleFromPb(in.Article))
	msg := ""
	if err != nil {
		msg = err.Error()
	} else {
		if count > 0 {
			nsqutil.SendEsMessage(nsqutil.ActionArticleModified, int64(id))
		}
	}
	fmt.Printf("duration: %v\n", time.Now().Sub(now))
	return &pb.ArticleUpdateResponse{
		Id:  count,
		Msg: msg,
	}, nil
}

// ArticleDelete
func (s *BlogDataServer) ArticleDelete(ctx context.Context, in *pb.ArticleDeleteRequest) (*pb.ArticleDeleteResponse, error) {
	now := time.Now()
	fmt.Printf("ArticleDelete: %s\n", now.Format(TIME_FORMAT))
	id := in.Id
	_, count, err := models.DeleteArticle(int(in.Id))
	msg := ""
	if err != nil {
		msg = err.Error()
	} else {
		nsqutil.SendEsMessage(nsqutil.ActionArticleRemoved, id)
	}
	fmt.Printf("duration: %v\n", time.Now().Sub(now))
	return &pb.ArticleDeleteResponse{
		Id:       id,
		TotalNum: int64(count),
		Msg:      msg,
	}, nil
}

4.写一个消费者

增加单个文档更新的方法

grpcdemo\models\esutils.go

package models

import (
	"context"
	"encoding/json"
	"fmt"
	"reflect"

	"github.com/olivere/elastic"
)

type ArticleSE struct {
	Id         int    `json:"id"`
	Title      string `json:"title"`
	Tags       string `json:"tags"`
	Short      string `json:"short"`
	Content    string `json:"content"`
	Author     string `json:"author"`
	Createtime int64  `json:"create_time"`
}

func (o *Article) ToSearchEntity() *ArticleSE {
	return &ArticleSE{
		Id:         o.Id,
		Title:      o.Title,
		Tags:       o.Tags,
		Short:      o.Short,
		Content:    o.Content,
		Author:     o.Author,
		Createtime: o.Createtime,
	}
}

func ArticleFromSearchEntity(value *ArticleSE) *Article {
	data := &Article{
		Id:         value.Id,
		Title:      value.Title,
		Tags:       value.Tags,
		Short:      value.Short,
		Content:    value.Content,
		Author:     value.Author,
		Createtime: value.Createtime,
	}
	return data
}

// 搜索相关
func SearchArticleBuildOne(ctx context.Context, item *Article, client *elastic.Client) {
	fmt.Printf("SearchArticleBuildOne: %d", item.Id)
	if item != nil {
		fmt.Printf("SearchArticleBuildOne: %d", item.Id)
		se_index := "blogweb_article"
		se_type := "doc"
		se_id := fmt.Sprintf("%d", item.Id)
		e := item.ToSearchEntity()

		put1, err := client.Index().
			Index(se_index).
			Type(se_type).
			Id(se_id).
			BodyJson(e).
			Do(ctx)
		if err != nil {
			// Handle error
			fmt.Println(err.Error())
		}
		fmt.Printf("Indexed %s to index %s, type %s\n", put1.Id, put1.Index, put1.Type)
	}
}

func SearchArticleRemoveOne(id int64) {
	fmt.Printf("SearchArticleRemoveOne: %d", id)
	client, err := elastic.NewSimpleClient(elastic.SetURL("http://127.0.0.1:9200/"))
	if err != nil {
		// Handle error
		fmt.Println(err.Error())
		return
	}
	defer client.Stop()

	ctx := context.Background()
	se_index := "blogweb_article"
	se_type := "doc"
	se_id := fmt.Sprintf("%d", id)
	res, err := client.Delete().
		Index(se_index).
		Type(se_type).
		Id(se_id).
		Do(ctx)
	if err != nil {
		// Handle error
		fmt.Println(err.Error())
	} else {
		fmt.Printf("delete result: %d, %s", res.Status, res.Result)
	}
}

func DumpQuery(src interface{}, err error) {
	if err != nil {
		panic(err)
	}
	data, err := json.MarshalIndent(src, "", "  ")
	if err != nil {
		panic(err)
	}
	fmt.Println(string(data))

}

func ArticleSearch(query string) []*Article {
	fmt.Printf("ArticleSearch: %s\n", query)
	var err error
	client, err := elastic.NewSimpleClient(elastic.SetURL("http://127.0.0.1:9200/"))
	if err != nil {
		// Handle error
		fmt.Println(err.Error())
		return nil
	}
	defer client.Stop()

	se_index := "blogweb_article"
	se_type := "doc"

	//条件查询
	formatS := `{
  "multi_match": {
      "query":    "%s",
      "fields":   [ "short", "content"]
  }
}`
	rawQ := elastic.NewRawStringQuery(
		fmt.Sprintf(formatS, query))
	DumpQuery(rawQ.Source())
	searchResult, err := client.Search(se_index).Type(se_type).
		Query(rawQ).
		From(0).Size(30).
		Do(context.Background())
	if err != nil {
		// Handle error
		fmt.Println(err.Error())
		return nil
	}
	fmt.Printf("Query took %d milliseconds. total: %d\n", searchResult.TookInMillis, searchResult.Hits.TotalHits)

	result := []*Article{}
	var ttyp ArticleSE
	for _, item := range searchResult.Each(reflect.TypeOf(ttyp)) {
		if t, ok := item.(ArticleSE); ok {
			result = append(result, ArticleFromSearchEntity(&t))
		}
	}
	return result
}

// 更新一个ES文档
func SearchArticleUpdate(id int) {
	fmt.Printf("SearchArticleUpdate: %d", id)
	ctx := context.Background()
	client, err := elastic.NewSimpleClient(elastic.SetURL("http://127.0.0.1:9200/"))
	if err != nil {
		// Handle error
		fmt.Println(err.Error())
		return
	}
	defer client.Stop()
	item := QueryArticleWithId(id)
	fmt.Printf("SearchArticleUpdate: %d\n", item.Id)
	se_index := "blogweb_article"
	se_type := "doc"
	se_id := fmt.Sprintf("%d", item.Id)
	e := item.ToSearchEntity()

	put1, err := client.Index().
		Index(se_index).
		Type(se_type).
		Id(se_id).
		BodyJson(e).
		Do(ctx)
	if err != nil {
		// Handle error
		fmt.Println(err.Error())
	}
	fmt.Printf("Indexed %s to index %s, type %s\n", put1.Id, put1.Index, put1.Type)

}

grpcdemo\nsqconsumer\main.go

package main

import (
	"blogweb_gin/database"
	"encoding/json"
	"fmt"
	"os"
	"os/signal"
	"syscall"
	"time"

	"grpcdemo/models"
	"grpcdemo/nsqutil"

	"github.com/nsqio/go-nsq"
)

// 消费者
type ConsumerEsT struct{}

// 处理消息
func (*ConsumerEsT) HandleMessage(msg *nsq.Message) error {
	fmt.Println("receive", msg.NSQDAddress, "message:", string(msg.Body))
	var data nsqutil.MessageEs
	err := json.Unmarshal([]byte(msg.Body), &data)
	if err != nil {
		return err
	}

	switch data.Action {
	case nsqutil.ActionArticleModified:
		models.SearchArticleUpdate(int(data.Id))
	case nsqutil.ActionArticleRemoved:
		models.SearchArticleRemoveOne(data.Id)
	default:
		fmt.Errorf("unknow queue action: %d", data.Action)

	}

	return nil
}

// 初始化消费者
func InitConsumer(address string) {
	cfg := nsq.NewConfig()
	cfg.LookupdPollInterval = time.Second                                                        //设置重连时间
	c1, err := nsq.NewConsumer(nsqutil.TopicEs, fmt.Sprintf("%s-channel", nsqutil.TopicEs), cfg) // 新建一个消费者
	if err != nil {
		panic(err)
	}
	c1.SetLogger(nil, 0)          //屏蔽系统日志
	c1.AddHandler(&ConsumerEsT{}) // 添加消费者接口

	// 建立NSQLookupd连接
	if err := c1.ConnectToNSQLookupd(address); err != nil {
		panic(err)
	}

	// if err := c1.ConnectToNSQD(address); err != nil {
	// 	panic(err)
	// }
}

func main() {
	database.InitMysql()
	InitConsumer("127.0.0.1:4161")
	// InitConsumer("127.0.0.1:4150")
	ch := make(chan os.Signal, 1)
	signal.Notify(ch, syscall.SIGTERM, syscall.SIGINT, syscall.SIGKILL, syscall.SIGHUP, syscall.SIGQUIT)
	<-ch
}

5.运行所有的服务

启动etcd

.\etcd.exe

启动 nsq

bin\nsqlookupd.exe

bin\nsqd --lookupd-tcp-address=127.0.0.1:4160

bin\nsqadmin --lookupd-http-address=127.0.0.1:4161

启动消费者

.\nsqconsumer.exe

启动gRPC数据库服务

D:\mygo\src\grpcdemo\server> .\server.exe

启动ES

D:\elasticsearch-6.7.1>bin\elasticsearch.bat

D:\kibana-6.7.1-windows-x86_64>bin\kibana.bat

查看索引列表

http://localhost:9200/_cat/indices

进入kibana

http://localhost:5601

删除旧索引

DELETE /blogweb_article

重建mapping

PUT /blogweb_article
{
    "settings": {
        "number_of_shards" :   1,
        "number_of_replicas" : 0
    },
  "mappings": {
      "doc" : {
        "properties" : {
          "author" : {
            "type": "text",
            "analyzer": "ik_max_word",
            "search_analyzer": "ik_max_word"
          },
          "content" : {
            "type": "text",
            "analyzer": "ik_max_word",
            "search_analyzer": "ik_max_word"
          },
          "create_time" : {
            "type" : "long"
          },
          "id" : {
            "type" : "long"
          },
          "short" : {
            "type": "text",
            "analyzer": "ik_max_word",
            "search_analyzer": "ik_max_word"
          },
          "tags" : {
            "type": "text",
            "analyzer": "ik_max_word",
            "search_analyzer": "ik_max_word"
          },
          "title" : {
            "type": "text",
            "analyzer": "ik_max_word",
            "search_analyzer": "ik_max_word"
          }
        }
      }
    }
}

运行重建索引数据的工具

D:\mygo\src\grpcdemo\es> .\es.exe

在kibana查看数据

GET /blogweb_article/doc/_search

启动blogweb

D:\mygo\src\blogweb_gin> .\blogweb_gin.exe

访问浏览器

http://localhost:8082/

登录webdemo

6.如何设计topic

1.按照结果

2.按照事件

推荐阅读