项目结构：
/*
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Push & Pull Pattern（推 - 拉）模式
# Author : geovindu,Geovin Du 涂聚文.
# IDE : goLang 2024.3.6 go 26.2
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/6/25 21:25
# User : geovindu
# Product : GoLand
# Project : godesginpattern
# File : config.go
*/
package config
import "time"
// ZMQ 端口 go get github.com/zeromq/goczmq
const (
PortRawMaterial = 5555
PortProcess = 5556
PortQuality = 5557
PortSale = 5558
)
// 超时毫秒
const ZmqTimeoutMs = 3000
// 消息分隔符
const (
MsgSep = "|"
KvSep = ":"
)
// 连接地址
const (
LocalTcpAddr = "tcp://localhost"
BindAddr = "tcp://*"
)
// 珠宝品类常量
var JewelryCategory = []string{
"钻石戒指", "黄金项链", "翡翠手镯", "铂金耳钉", "彩宝吊坠",
}
// 质检等级
var QualityGrade = []string{
"S级(收藏)", "A级(精品)", "B级(常规)", "C级(特价)",
}
// 休眠间隔
const (
SleepRaw = 2 * time.Second
SleepProcess = 1 * time.Second
SleepQuality = 500 * time.Millisecond
SleepSale = 1 * time.Second
IdleSleep = 100 * time.Millisecond // 无消息空转休眠
)
/*
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Push & Pull Pattern（推 - 拉）模式
# Author : geovindu,Geovin Du 涂聚文.
# IDE : goLang 2024.3.6 go 26.2
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/6/25 21:26
# User : geovindu
# Product : GoLand
# Project : godesginpattern
# File : logger.go
*/
package utils
import (
"fmt"
"log"
"os"
"time"
)
func GetLogger(module string) *log.Logger {
prefix := fmt.Sprintf("[%s] ", module)
logger := log.New(os.Stdout, prefix, log.LstdFlags)
return logger
}
// 格式化时间输出辅助
func NowStr() string {
return time.Now().Format("2006-01-02 15:04:05")
}
/*
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Push & Pull Pattern（推 - 拉）模式
# Author : geovindu,Geovin Du 涂聚文.
# IDE : goLang 2024.3.6 go 26.2
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/6/25 21:26
# User : geovindu
# Product : GoLand
# Project : godesginpattern
# File : message.go
*/
package model
import (
"godesginpattern/pushpull/config"
"strings"
)
// BaseMessage 统一消息打包解包
type BaseMessage struct{}
// Pack map 转消息字符串
// 首字母大写，跨包可访问
var MsgHelper = &BaseMessage{}
func (b *BaseMessage) Pack(data map[string]string) string {
var parts []string
for k, v := range data {
parts = append(parts, k+config.KvSep+v)
}
return strings.Join(parts, config.MsgSep)
}
func (b *BaseMessage) Unpack(raw string) map[string]string {
res := make(map[string]string)
items := strings.Split(raw, config.MsgSep)
for _, item := range items {
kv := strings.SplitN(item, config.KvSep, 2)
if len(kv) == 2 {
res[kv[0]] = kv[1]
}
}
return res
}
/*
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Push & Pull Pattern（推 - 拉）模式
# Author : geovindu,Geovin Du 涂聚文.
# IDE : goLang 2024.3.6 go 26.2
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/6/25 21:25
# User : geovindu
# Product : GoLand
# Project : godesginpattern
# File : socket_factory.go
*/
package core
import (
"bufio"
"fmt"
"godesginpattern/pushpull/config"
"godesginpattern/pushpull/utils"
"io"
"net"
"sync"
"time"
)
var logger = utils.GetLogger("SocketFactory")
type TcpPushSocket struct {
listener net.Listener
clients []net.Conn
mu sync.Mutex
}
func NewTcpPushSocket(port int) *TcpPushSocket {
listener, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
if err != nil {
logger.Fatalf("创建Push监听失败: %v", err)
}
sock := &TcpPushSocket{listener: listener}
go sock.acceptLoop()
logger.Printf("Push 绑定成功: tcp://localhost:%d", port)
return sock
}
func (s *TcpPushSocket) acceptLoop() {
for {
conn, err := s.listener.Accept()
if err != nil {
logger.Printf("Accept错误: %v", err)
return
}
s.mu.Lock()
s.clients = append(s.clients, conn)
s.mu.Unlock()
logger.Printf("新客户端连接: %s", conn.RemoteAddr())
}
}
func (s *TcpPushSocket) Send(msg string) (int, error) {
s.mu.Lock()
defer s.mu.Unlock()
for _, conn := range s.clients {
conn.SetWriteDeadline(time.Now().Add(time.Duration(config.ZmqTimeoutMs) * time.Millisecond))
if _, err := conn.Write([]byte(msg + "\n")); err != nil {
logger.Printf("发送失败: %v", err)
}
}
return len(msg), nil
}
func (s *TcpPushSocket) Close() {
s.mu.Lock()
defer s.mu.Unlock()
for _, conn := range s.clients {
conn.Close()
}
s.listener.Close()
}
type TcpPullSocket struct {
conn net.Conn
reader *bufio.Reader
}
func NewTcpPullSocket(port int) *TcpPullSocket {
conn, err := net.Dial("tcp", fmt.Sprintf("localhost:%d", port))
if err != nil {
logger.Fatalf("连接Push失败: %v", err)
}
logger.Printf("Pull 连接成功: tcp://localhost:%d", port)
return &TcpPullSocket{
conn: conn,
reader: bufio.NewReader(conn),
}
}
var ErrConnectionClosed = fmt.Errorf("connection closed")
func (s *TcpPullSocket) Recv(nonBlocking bool) (string, error) {
if nonBlocking {
s.conn.SetReadDeadline(time.Now().Add(50 * time.Millisecond))
} else {
s.conn.SetReadDeadline(time.Now().Add(time.Duration(config.ZmqTimeoutMs) * time.Millisecond))
}
msg, err := s.reader.ReadString('\n')
if err != nil {
if nonBlocking && (err == bufio.ErrBufferFull || isTimeoutError(err)) {
return "", fmt.Errorf("resource temporarily unavailable")
}
if err == io.EOF {
return "", ErrConnectionClosed
}
return "", err
}
return msg[:len(msg)-1], nil
}
func (s *TcpPullSocket) Close() {
s.conn.Close()
}
func isTimeoutError(err error) bool {
if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
return true
}
return false
}
/*
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Push & Pull Pattern（推 - 拉）模式
# Author : geovindu,Geovin Du 涂聚文.
# IDE : goLang 2024.3.6 go 26.2
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/6/25 21:26
# User : geovindu
# Product : GoLand
# Project : godesginpattern
# File : rawmaterial.go
*/
package service
import (
"godesginpattern/pushpull/config"
"godesginpattern/pushpull/core"
"godesginpattern/pushpull/model"
"godesginpattern/pushpull/utils"
"math/rand"
"strconv"
"time"
)
var rawLog = utils.GetLogger("RawMaterialService")
type RawMaterialService struct {
pushSock *core.TcpPushSocket
}
func NewRawMaterialService() *RawMaterialService {
return &RawMaterialService{
pushSock: core.NewTcpPushSocket(config.PortRawMaterial),
}
}
func (r *RawMaterialService) RunProduce(total int) {
rawLog.Println("原料采购服务启动")
rand.Seed(time.Now().UnixNano())
for i := 1; i <= total; i++ {
cat := config.JewelryCategory[rand.Intn(len(config.JewelryCategory))]
msg := model.MsgHelper.Pack(map[string]string{
"type": "原料订单",
"order_id": "RAW_" + strconv.Itoa(i),
"category": cat,
"status": "已采购待加工",
})
_, _ = r.pushSock.Send(msg)
rawLog.Printf("推送: %s", msg)
time.Sleep(config.SleepRaw)
}
rawLog.Println("原料订单全部推送完成")
time.Sleep(time.Second)
r.pushSock.Close()
}
/*
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Push & Pull Pattern（推 - 拉）模式
# Author : geovindu,Geovin Du 涂聚文.
# IDE : goLang 2024.3.6 go 26.2
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/6/25 21:26
# User : geovindu
# Product : GoLand
# Project : godesginpattern
# File : process.go
*/
package service
import (
"godesginpattern/pushpull/config"
"godesginpattern/pushpull/core"
"godesginpattern/pushpull/model"
"godesginpattern/pushpull/utils"
"math/rand"
"strings"
"time"
)
var procLog = utils.GetLogger("Process")
type ProcessService struct {
pullSock *core.TcpPullSocket
pushSock *core.TcpPushSocket
done chan struct{}
}
func NewProcessService() *ProcessService {
return &ProcessService{
pullSock: core.NewTcpPullSocket(config.PortRawMaterial),
pushSock: core.NewTcpPushSocket(config.PortProcess),
done: make(chan struct{}),
}
}
func (p *ProcessService) RunPipeline() {
procLog.Println("加工车间启动")
for {
select {
case <-p.done:
procLog.Println("加工车间停止")
return
default:
msg, err := p.pullSock.Recv(true)
if err != nil {
if err == core.ErrConnectionClosed {
procLog.Println("加工车间连接关闭")
return
}
if err.Error() == "resource temporarily unavailable" {
time.Sleep(config.IdleSleep)
continue
}
procLog.Printf("接收错误: %v", err)
time.Sleep(config.IdleSleep)
continue
}
data := model.MsgHelper.Unpack(msg)
procLog.Printf("收到: %s", msg)
time.Sleep(config.SleepProcess + time.Duration(rand.Intn(2000))*time.Millisecond)
sendMsg := model.MsgHelper.Pack(map[string]string{
"type": "成品珠宝",
"order_id": strings.Replace(data["order_id"], "RAW", "FIN", 1),
"category": data["category"],
"status": "已加工待质检",
})
_, _ = p.pushSock.Send(sendMsg)
procLog.Printf("加工完成: %s", sendMsg)
}
}
}
func (p *ProcessService) Stop() {
close(p.done)
p.pullSock.Close()
p.pushSock.Close()
}
/*
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Push & Pull Pattern（推 - 拉）模式
# Author : geovindu,Geovin Du 涂聚文.
# IDE : goLang 2024.3.6 go 26.2
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/6/25 21:26
# User : geovindu
# Product : GoLand
# Project : godesginpattern
# File : quality.go
*/
package service
import (
"godesginpattern/pushpull/config"
"godesginpattern/pushpull/core"
"godesginpattern/pushpull/model"
"godesginpattern/pushpull/utils"
"math/rand"
"time"
)
var qualityLog = utils.GetLogger("Quality")
type QualityService struct {
pullSock *core.TcpPullSocket
pushSock *core.TcpPushSocket
done chan struct{}
}
func NewQualityService() *QualityService {
return &QualityService{
pullSock: core.NewTcpPullSocket(config.PortProcess),
pushSock: core.NewTcpPushSocket(config.PortQuality),
done: make(chan struct{}),
}
}
func (q *QualityService) RunPipeline() {
qualityLog.Println("质检中心启动")
for {
select {
case <-q.done:
qualityLog.Println("质检中心停止")
return
default:
msg, err := q.pullSock.Recv(true)
if err != nil {
if err == core.ErrConnectionClosed {
qualityLog.Println("质检中心连接关闭")
return
}
if err.Error() == "resource temporarily unavailable" {
time.Sleep(config.IdleSleep)
continue
}
qualityLog.Printf("接收错误: %v", err)
time.Sleep(config.IdleSleep)
continue
}
data := model.MsgHelper.Unpack(msg)
qualityLog.Printf("收到: %s", msg)
time.Sleep(config.SleepQuality + time.Duration(rand.Intn(1500))*time.Millisecond)
sendMsg := model.MsgHelper.Pack(map[string]string{
"type": "质检成品",
"order_id": data["order_id"],
"category": data["category"],
"grade": config.QualityGrade[rand.Intn(len(config.QualityGrade))],
"status": "可销售",
})
_, _ = q.pushSock.Send(sendMsg)
qualityLog.Printf("质检完成: %s", sendMsg)
}
}
}
func (q *QualityService) Stop() {
close(q.done)
q.pullSock.Close()
q.pushSock.Close()
}
/*
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Push & Pull Pattern（推 - 拉）模式
# Author : geovindu,Geovin Du 涂聚文.
# IDE : goLang 2024.3.6 go 26.2
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/6/25 21:26
# User : geovindu
# Product : GoLand
# Project : godesginpattern
# File : sale.go
*/
package service
import (
"godesginpattern/pushpull/config"
"godesginpattern/pushpull/core"
"godesginpattern/pushpull/model"
"godesginpattern/pushpull/utils"
"math/rand"
"strconv"
"strings"
"time"
)
var saleLog = utils.GetLogger("Sale")
type SaleService struct {
pullSock *core.TcpPullSocket
pushSock *core.TcpPushSocket
done chan struct{}
}
func NewSaleService() *SaleService {
return &SaleService{
pullSock: core.NewTcpPullSocket(config.PortQuality),
pushSock: core.NewTcpPushSocket(config.PortSale),
done: make(chan struct{}),
}
}
func (s *SaleService) RunPipeline() {
saleLog.Println("销售部启动")
for {
select {
case <-s.done:
saleLog.Println("销售部停止")
return
default:
msg, err := s.pullSock.Recv(true)
if err != nil {
if err == core.ErrConnectionClosed {
saleLog.Println("销售部连接关闭")
return
}
if err.Error() == "resource temporarily unavailable" {
time.Sleep(config.IdleSleep)
continue
}
saleLog.Printf("接收错误: %v", err)
time.Sleep(config.IdleSleep)
continue
}
data := model.MsgHelper.Unpack(msg)
saleLog.Printf("收到: %s", msg)
time.Sleep(config.SleepSale + time.Duration(rand.Intn(1500))*time.Millisecond)
sendMsg := model.MsgHelper.Pack(map[string]string{
"type": "销售完成",
"order_id": strings.Replace(data["order_id"], "FIN", "SALE", 1),
"category": data["category"],
"price": strconv.Itoa(rand.Intn(49000)+1000) + "元",
"status": "已售出",
})
_, _ = s.pushSock.Send(sendMsg)
saleLog.Printf("销售完成: %s", sendMsg)
}
}
}
func (s *SaleService) Stop() {
close(s.done)
s.pullSock.Close()
s.pushSock.Close()
}
/*
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Push & Pull Pattern（推 - 拉）模式
# Author : geovindu,Geovin Du 涂聚文.
# IDE : goLang 2024.3.6 go 26.2
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/6/25 21:26
# User : geovindu
# Product : GoLand
# Project : godesginpattern
# File : aftersale.go
*/
package service
import (
"godesginpattern/pushpull/config"
"godesginpattern/pushpull/core"
"godesginpattern/pushpull/model"
"godesginpattern/pushpull/utils"
"time"
)
var afterLog = utils.GetLogger("AfterSale")
type AfterSaleService struct {
pullSock *core.TcpPullSocket
done chan struct{}
}
func NewAfterSaleService() *AfterSaleService {
return &AfterSaleService{
pullSock: core.NewTcpPullSocket(config.PortSale),
done: make(chan struct{}),
}
}
func (a *AfterSaleService) RunConsumer() {
afterLog.Println("售后维保启动")
for {
select {
case <-a.done:
afterLog.Println("售后维保停止")
return
default:
msg, err := a.pullSock.Recv(true)
if err != nil {
if err == core.ErrConnectionClosed {
afterLog.Println("售后维保连接关闭")
return
}
if err.Error() == "resource temporarily unavailable" {
time.Sleep(config.IdleSleep)
continue
}
afterLog.Printf("接收错误: %v", err)
time.Sleep(config.IdleSleep)
continue
}
data := model.MsgHelper.Unpack(msg)
afterLog.Printf("收到单据: %s", msg)
afterLog.Printf("✅ 维保生效 | 品类:%s | 订单:%s",
data["category"], data["order_id"])
}
}
}
func (a *AfterSaleService) Stop() {
close(a.done)
a.pullSock.Close()
}
调用：
/*
# 版权所有 2026 ©涂聚文有限公司™ ®
# 许可信息查看：言語成了邀功盡責的功臣，還需要行爲每日來值班嗎
# 描述：Push & Pull Pattern（推 - 拉）模式
# Author : geovindu,Geovin Du 涂聚文.
# IDE : goLang 2024.3.6 go 26.2
# os : windows 10
# database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j
# Datetime : 2026/6/25 21:26
# User : geovindu
# Product : GoLand
# Project : godesginpattern
# File : pushpullbll.go
*/
package bll
import (
"godesginpattern/pushpull/service"
"godesginpattern/pushpull/utils"
"os"
"os/signal"
"syscall"
"time"
)
var ppLog = utils.GetLogger("pushpullbll")
func PushPullMain() {
sig := make(chan os.Signal, 1)
signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
raw := service.NewRawMaterialService()
proc := service.NewProcessService()
quality := service.NewQualityService()
sale := service.NewSaleService()
after := service.NewAfterSaleService()
go func() {
<-sig
ppLog.Println("退出中...")
proc.Stop()
quality.Stop()
sale.Stop()
after.Stop()
os.Exit(0)
}()
go proc.RunPipeline()
time.Sleep(200 * time.Millisecond)
go quality.RunPipeline()
time.Sleep(200 * time.Millisecond)
go sale.RunPipeline()
time.Sleep(200 * time.Millisecond)
go after.RunConsumer()
time.Sleep(200 * time.Millisecond)
ppLog.Println("✅ 珠宝 Push-Pull 流水线启动完成")
raw.RunProduce(5)
time.Sleep(5 * time.Second)
proc.Stop()
quality.Stop()
sale.Stop()
after.Stop()
ppLog.Println("✅ 珠宝 Push-Pull 流水线运行结束")
}
输出：
哲学管理(学)人生, 文学艺术生活, 自动(计算机学)物理(学)工作, 生物(学)化学逆境, 历史(学)测绘(学)时间, 经济(学)数学金钱(理财), 心理(学)医学情绪, 诗词美容情感, 美学建筑(学)家园, 解构建构(分析)整合学习, 智商情商(IQ、EQ)运筹(学)生存.---Geovin Du(涂聚文）
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。