1 前置条件
Golang基本情况自行baidu/google
1.1 环境与工具
IDE：liteide (windows )、mingw-w64 (gcc)
DB：SQL Server 2008 R2
MQ: Kakfa
1.2 环境变量
Golang 需要将GOPATH添加到环境变量中
1.3 版本查看
Cmd中输入 go version 可以查看当前golang版本
2 示例程序功能简介
2.1 主功能
- 查询数据表A获取记录的消息编号MSG-Id.
- 从数据表B查询大于MSG-Id的100条连续数据.
- 将查询到的100条数据依次序列化成JSON数据，推送到消息队列Kafka中（供其它软件使用）.
- 100条推送成功后将最大的MSG-Id更新到数据表A.
2.2 辅助项
- 循环执行，每执行主功能一次，挂起2秒
- 记录关键日志到文件
- 从配置文件读取信息
- 在Console上显示实时信息
3 代码描述
3.1 目录结构与程序包
MainExe：主项目模块的目录
Domain领域层目录
xxDomain.go 领域层源码文件（Package yyDomain）
Model模型层目录
xxModel.go 模型层源码文件（Package yyModel）
Main.go:主函数源码文件，Package Main
Common：通用/公共功能模块目录
MQ_kafka帮助类子目录
Sarama.go: kafka帮助类源码（package KafkaHelper）
DB_SQLServer帮助类子目录
SQLServer.go: SQLServer帮助类源码（package SQLHelper）
File_CFG帮助类子目录
CFG.go: 配置文件操作帮助类源码（package CFGHelper）
在Goalng中源码的物理文件是按照目录（Directory）来管理, 每个*.go文件都存放在一个目录下，一个目录下的所有源文件只能属于相同的包package。在逻辑上，源码使用包（package）这种语法元素来组织源码，所有语法可见性均定义在package这个级别，每个*.go文件都属于一个Package。
导入：Import（”目录/子目录”）
程序调用: PackageName.xxxMethod/yyyFunction
3.2 源码文件SQLServer.go
这里用到了go-mssqldb这个包，在命令行cmd使用go get github.com/denisenkom/go-mssqldb 可以将程序包下载到本地
这个包在import时，前面加了横线，说明即使用【import _ 包路径】只是引用该包，仅仅是为了调用init()函数，所以无法通过包名来调用包中的其他函数。
package SQLHelper
import (
"Common/File_CFG"
"database/sql"
"encoding/json"
"fmt"
"log"
"time"
_ "github.com/denisenkom/go-mssqldb"
)
func QueryDataReturnString(strSQL string /*, connInstance *sql.DB*/) (strValue string, err error) {
ip, port, user, password, database, err := Base_ConfigFileHelper.GetSQLServerConfig()
if err != nil {
log.Fatal("Read Config File failed:", err.Error())
}
connString := fmt.Sprintf("server=%s;port%s;database=%s;user id=%s;password=%s", ip, port, database, user, password)
connInstance, err := sql.Open("mssql", connString)
if err != nil {
log.Fatal("Open Connection failed:", err.Error())
}
defer connInstance.Close()
var strLastId string
errTemp := connInstance.QueryRow(strSQL).Scan(&strLastId)
if errTemp != nil {
log.Fatal("Query failed:", errTemp.Error())
}
return strLastId, errTemp
}
func QueryDataReturnArray(strSQL string /*, connInstance *sql.DB*/) (dataRows *sql.Rows) {
ip, port, user, password, database, err := Base_ConfigFileHelper.GetSQLServerConfig()
if err != nil {
log.Fatal("Read Config File failed:", err.Error())
}
connString := fmt.Sprintf("server=%s;port%d;database=%s;user id=%s;password=%s", ip, port, database, user, password)
// if isdebug {
// fmt.Println(connString)
// }
connInstance, err := sql.Open("mssql", connString)
if err != nil {
log.Fatal("Open Connection failed:", err.Error())
}
defer connInstance.Close()
stmt, err := connInstance.Prepare(strSQL)
if err != nil {
log.Fatal("Prepare failed:", err.Error())
}
rowsArray, err := stmt.Query()
if err != nil {
log.Fatal("Query failed:", err.Error())
}
stmt.Close()
//defer rowsArray.Close()
return rowsArray
}
func ExeNoQuery(strSQL string /*, connInstance *sql.DB*/) (returnResult int64, err error) {
ip, port, user, password, database, err := Base_ConfigFileHelper.GetSQLServerConfig()
if err != nil {
log.Fatal("Read Config File failed:", err.Error())
}
connString := fmt.Sprintf("server=%s;port%d;database=%s;user id=%s;password=%s", ip, port, database, user, password)
connInstance, err := sql.Open("mssql", connString)
if err != nil {
log.Fatal("Open Connection failed:", err.Error())
}
defer connInstance.Close()
stmt, err := connInstance.Prepare(strSQL)
if err != nil {
log.Fatal("Prepare failed:", err.Error())
}
defer stmt.Close()
returnRows, err := stmt.Exec()
if err != nil {
log.Fatal("Query failed:", err.Error())
}
//我们可以获得插入的id
//id, err := rs.LastInsertId()
//可以获得影响行数
result, err := returnRows.RowsAffected()
return result, err
}
// 参考资料
//Go-SQLServer https://www.cnblogs.com/dfsxh/p/10207866.html
//Go-SQLServer https://www.cnblogs.com/Liang2790912648/p/10604310.html
//Go-SQLServer https://www.jianshu.com/p/136ac470fc17
//Go-SQLServer https://blog.csdn.net/chen_peng7/article/details/89317922
//GO-MySQL(scan) https://zhidao.baidu.com/question/1759806412743140028.html
3.3 源码文件Sarama.go
package KafkaHelper
import (
"fmt"
"time"
"github.com/Shopify/sarama"
)
func CreateInstance() (syncClient sarama.SyncProducer, err error) {
//初始化配置
newConfig := sarama.NewConfig()
newConfig.Producer.RequiredAcks = sarama.WaitForAll
newConfig.Producer.Partitioner = sarama.NewRandomPartitioner
newConfig.Producer.Return.Successes = true
//生产者
client, err := sarama.NewSyncProducer([]string{"172.16.77.53:9092"}, newConfig)
if err != nil {
fmt.Println("producer close,err:", err)
return nil, err
} else {
return client, err
}
}
func SendMessage(syncClient sarama.SyncProducer, InMessage string) {
msg := &sarama.ProducerMessage{}
msg.Topic = "02.dbo "
msg.Value = sarama.StringEncoder(InMessage)
//发送消息
pid, offset, err := syncClient.SendMessage(msg)
if err != nil {
fmt.Println(time.Now().Format("2006-01-02 15:04:05")+" send failed: ", err)
} else {
fmt.Printf(time.Now().Format("2006-01-02 15:04:05")+" send success:1 pid:%v offset:%v\n", pid, offset)
}
}
因为在.net core 环境下用Confluent成功，本想goalng用Confluent package继续测试，但是下载这个包很困难（墙），所以放弃了。
// 参考资料
//Go-Kafka https://www.cnblogs.com/vincenshen/p/9824486.html
//Go-Kafka https://www.cnblogs.com/pyyu/p/8371649.html
//Go-Kafka https://blog.csdn.net/lanyang123456/article/details/78377152
//Go-Kafka https://www.jianshu.com/p/1136f37cc419
3.4 源码文件CFG.go
package CFGHelper
import (
"fmt"
"github.com/Unknwon/goconfig"
)
func GetSQLServerConfig() (ip string, port string, username string, password string, dbname string, err error) {
cfgFile, err := goconfig.LoadConfigFile("conf.ini")
if err != nil {
fmt.Println(err)
}
ipGet, err := cfgFile.GetValue("DB_SQLServer", "ip")
if err != nil {
fmt.Println(err)
}
portGet, err := cfgFile.GetValue("DB_SQLServer", "port")
if err != nil {
fmt.Println(err)
}
usernameGet, err := cfgFile.GetValue("DB_SQLServer", "username")
if err != nil {
fmt.Println(err)
}
passwordGet, err := cfgFile.GetValue("DB_SQLServer", "password")
if err != nil {
fmt.Println(err)
}
dbnameGet, err := cfgFile.GetValue("DB_SQLServer", "dbname")
if err != nil {
fmt.Println(err)
}
return ipGet, portGet, usernameGet, passwordGet, dbnameGet, err
}
func GetKafkaConfig() (ip string, topic string, groupid string, err error) {
cfgFile, err := goconfig.LoadConfigFile("../conf.ini")
if err != nil {
fmt.Println(err)
}
ipGet, err := cfgFile.GetValue("MQ_Kafka", "ip")
if err != nil {
fmt.Println(err)
}
topicGet, err := cfgFile.GetValue("MQ_Kafka", "topic")
if err != nil {
fmt.Println(err)
}
groupidGet, err := cfgFile.GetValue("MQ_Kafka", "groupid")
if err != nil {
fmt.Println(err)
}
return ipGet, topicGet, groupidGet, err
}
3.4 源码文件main.go
import 里面有很多测试导入的包，如果报错就屏蔽掉
package main
import (
"MainMenthod/Domain"
"fmt"
"io"
"net/http"
"os"
"reflect"
"time"
)
func main() {
mainDomain.DoWork_Init()
for true {
//var currentMonth：=time.Now().String()
mainDomain.DoWork_SendMSG()
time.Sleep(2 * time.Second)
}
mainDomain.DoWork_Clean()
//testReflect()
//testgoroutine()
//testhttpget()
fmt.Println("program exit")
}
3.4 源码文件xxModel.go
package yyModel
type BrushCardModel struct {
Odid int32
StepCode string
EmpCode string
SysCode string
Sort string
OrderCode string
LoadDate string
WorkHours int32
}
3.4 源码文件yyDomain.go
package mainDomain
import (
"CommonCode/DB_SQLServer"
"CommonCode/MQ_Kafka"
"MainMenthod/Model"
"database/sql"
"encoding/json"
"fmt"
"log"
"os"
"strconv"
"strings"
"time"
"github.com/Shopify/sarama"
_ "github.com/denisenkom/go-mssqldb"
)
var kafkaProducer sarama.SyncProducer
func DoWork_Init() {
producer, err := KafkaHelper.CreateInstance()
if err != nil {
log.Fatal("Create Kafka Producer failed:", err.Error())
fmt.Println("Create Kafka Producer failed" + err.Error())
return
} else {
kafkaProducer = producer
}
}
func DoWork_Clean() {
kafkaProducer.Close()
}
func DoWork_SendMSG() {
//固定的时间；“2006-01-02”
file := "./" + time.Now().Format("20060102") + ".txt"
logFile, err := os.OpenFile(file, os.O_RDWR|os.O_CREATE|os.O_APPEND, os.ModePerm)
if nil != err {
panic(err)
}
defer logFile.Close()
loger := log.New(logFile, "", log.Ldate|log.Ltime|log.Lshortfile)
var strSQL_LastId string = "select LastMSGId from A where MSGTable=‘B‘"
LastMSGIdOfBC, err := SQLHelper.QueryDataReturnString(strSQL_LastId /*, connInstance*/)
if err != nil {
fmt.Println("Query LastMSGId failed" + err.Error())
loger.Output(2, "Query LastMSGId failed"+err.Error()+"\r\n")
}
var builderSQL strings.Builder
builderSQL.WriteString(" select top 100 MaterialCardBrushID,Odid,SysCode,OrderCode,")
builderSQL.WriteString(" EmpCode,StepCode,Sort,WorkHours, ")
builderSQL.WriteString(" CONVERT(varchar(100), LoadDate, 20) ")
builderSQL.WriteString(" from B where MaterialCardBrushID >" + LastMSGIdOfBC)
builderSQL.WriteString(" order by MaterialCardBrushID")
//fmt.Println(builderSQL.String())
var rowsArray *sql.Rows
rowsArray = SQLHelper.QueryDataReturnArray(builderSQL.String() /*, connInstance*/)
var materialCardBrushID int
for rowsArray.Next() {
ins_BrushCard := &mainModel.BrushCardModel{}
err = rowsArray.Scan(&materialCardBrushID, &ins_BrushCard.Odid, &ins_BrushCard.SysCode, &ins_BrushCard.OrderCode,
&ins_BrushCard.EmpCode, &ins_BrushCard.StepCode, &ins_BrushCard.Sort, &ins_BrushCard.WorkHours,
&ins_BrushCard.LoadDate)
if err != nil {
fmt.Println("Get Data from Row failed" + err.Error())
loger.Output(2, "Get Data from Row failed"+err.Error()+"\r\n")
break
} else {
jsonOfBrushCard, err := json.Marshal(ins_BrushCard)
if err != nil {
fmt.Println("生成json字符串错误" + err.Error())
loger.Output(2, "生成json字符串错误"+err.Error()+"\r\n")
}
var strValue string = string(jsonOfBrushCard)
loger.Output(2, strValue+"\r\n")
KafkaHelper.SendMessage(kafkaProducer, strValue)
}
}
rowsArray.Close()
var strSQL_Update string = "update SendLastMSGFlag set LastMSGId=" + strconv.Itoa(materialCardBrushID) + " where MSGTable=‘B‘"
result, err := SQLHelper.ExeNoQuery(strSQL_Update /*, connInstance*/)
if err != nil {
//log.Fatal("Update failed:materialCardBrushID="+strconv.Itoa(materialCardBrushID), err.Error())
fmt.Println(time.Now().Format("2006-01-02 15:04:05") + " LastMSGId:" + strconv.Itoa(materialCardBrushID) + " failed:" + err.Error())
loger.Output(2, " LastMSGId:"+strconv.Itoa(materialCardBrushID)+" failed:"+err.Error()+"\r\n")
} else {
fmt.Println(time.Now().Format("2006-01-02 15:04:05") + " update success:" + strconv.FormatInt(result, 10) + " LastMSGId:" + strconv.Itoa(materialCardBrushID))
loger.Output(2, " update success:"+strconv.FormatInt(result, 10)+" LastMSGId:"+strconv.Itoa(materialCardBrushID)+"\r\n")
}
}
// 参考资料
// Go-JSON https://www.cnblogs.com/pyyu/p/8309917.html
// Go-JSON https://www.jianshu.com/p/3534532e06ed
// Go-JSON https://studygolang.com/articles/8526 SQL结果集-JSON
// Go-base https://www.jianshu.com/p/dbb39c69e3f1 类型转换string、int、int64
// Go-base https://www.cnblogs.com/smallleiit/p/10694481.html 类型转换string、int、int64
// Go-base https://www.cnblogs.com/pyyu/p/8258133.html 自定义类型struct
// Go-base https://studygolang.com/articles/3383 log简介
3.4 配置文件conf.ini
[DB_SQLServer] ip=127.0.0.1 port=1433 username=sa password=123456 dbname=test [MQ_Kafka] ip=127.0.0.1:9092 topic=topcoco groupid=zzc
对于熟悉一门编程语言，学习另外一种语言，总有一些困难：
- 首先是开始时的开发--编译-运行环境，也就是hello world，以及调试：需要搜集大量资料或教程才能敲门
- 其次是习惯，从面向过程转面向对象或者反之，都需要克服习惯
- 最后，是做一个工作上用到的，不大的有用的程序，除非你的兴趣特别强
原文地址：https://www.cnblogs.com/hhhh2010/p/11504305.html
时间： 2024-10-03 15:48:17