Skip to content

孙龙 / ichunt-micro-v2

Go to a project
Commit abcdca65 by 孙龙
Options

忽略文件

parent 527c92e4
Expand all
Inline Side-by-side
Showing with 4 additions and 529 deletions
.gitignore 0 → 100644
/go.sum
/cmd/*
/conf/dev/*
/conf/prod/*
cmd/bat.bat deleted 100644 → 0
SET CGO_ENABLED=0
SET GOOS=linux
SET GOARCH=amd64
cmd/ichunt-micro deleted 100644 → 0
The file could not be displayed because it is too large.
cmd/load_balance/main.go deleted 100644 → 0
package main
import (
"bytes"
"compress/gzip"
"context"
"fmt"
"ichunt-micro/proxy/load_balance"
"ichunt-micro/registry"
_ "ichunt-micro/registry/etcd"
"io/ioutil"
"log"
"net"
"net/http"
"net/http/httputil"
"net/url"
"strconv"
"strings"
"time"
)
var (
addr = "192.168.2.246:2002"
//addr = "192.168.2.246:2002"
transport = &http.Transport{
DialContext: (&net.Dialer{
Timeout: 30 * time.Second, //连接超时
KeepAlive: 30 * time.Second, //长连接超时时间
}).DialContext,
MaxIdleConns: 100, //最大空闲连接
IdleConnTimeout: 90 * time.Second, //空闲超时时间
TLSHandshakeTimeout: 10 * time.Second, //tls握手超时时间
ExpectContinueTimeout: 1 * time.Second, //100-continue状态码超时时间
}
)
func NewMultipleHostsReverseProxy() *httputil.ReverseProxy {
//请求协调者
director := func(req *http.Request) {
nextAddr, err := load_balance.LoadBalanceConfig.GetService(context.TODO(), "comment_service")
fmt.Println(nextAddr,err)
if err != nil {
log.Fatal("get next addr fail")
}
target, err := url.Parse("http://"+nextAddr)
if err != nil {
log.Fatal(err)
}
targetQuery := target.RawQuery
req.URL.Scheme = target.Scheme
req.URL.Host = target.Host
req.URL.Path = singleJoiningSlash(target.Path, req.URL.Path)
if targetQuery == "" || req.URL.RawQuery == "" {
req.URL.RawQuery = targetQuery + req.URL.RawQuery
} else {
req.URL.RawQuery = targetQuery + "&" + req.URL.RawQuery
}
if _, ok := req.Header["User-Agent"]; !ok {
req.Header.Set("User-Agent", "user-agent")
}
//只在第一代理中设置此header头
req.Header.Set("X-Real-Ip", req.RemoteAddr)
}
//更改内容
modifyFunc := func(resp *http.Response) error {
if strings.Contains(resp.Header.Get("Connection"), "Upgrade") {
return nil
}
var payload []byte
var readErr error
if strings.Contains(resp.Header.Get("Content-Encoding"), "gzip") {
gr, err := gzip.NewReader(resp.Body)
if err != nil {
return err
}
payload, readErr = ioutil.ReadAll(gr)
resp.Header.Del("Content-Encoding")
} else {
payload, readErr = ioutil.ReadAll(resp.Body)
}
if readErr != nil {
return readErr
}
//异常请求时设置StatusCode
if resp.StatusCode != 200 {
payload = []byte("StatusCode error:" + string(payload))
}
resp.Body = ioutil.NopCloser(bytes.NewBuffer(payload))
resp.ContentLength = int64(len(payload))
resp.Header.Set("Content-Length", strconv.FormatInt(int64(len(payload)), 10))
return nil
}
//错误回调 ：关闭real_server时测试，错误回调
//范围：transport.RoundTrip发生的错误、以及ModifyResponse发生的错误
errFunc := func(w http.ResponseWriter, r *http.Request, err error) {
//todo 如果是权重的负载则调整临时权重
http.Error(w, "ErrorHandler error:"+err.Error(), 500)
}
return &httputil.ReverseProxy{Director: director, Transport: transport, ModifyResponse: modifyFunc, ErrorHandler: errFunc}
}
func singleJoiningSlash(a, b string) string {
aslash := strings.HasSuffix(a, "/")
bslash := strings.HasPrefix(b, "/")
switch {
case aslash && bslash:
return a + b[1:]
case !aslash && !bslash:
return a + "/" + b
}
return a + b
}
func main() {
//初始化注册中心 注册etcd 服务中心
registryInst, err := registry.InitRegistry(context.TODO(), "etcd",
registry.WithAddrs([]string{"192.168.2.232:2379"}),
registry.WithTimeout(time.Second),
registry.WithRegistryPath("/ichuntMicroService/"),
registry.WithHeartBeat(5),
)
if err != nil {
fmt.Println(err)
fmt.Printf("init registry failed, err:%v", err)
return
}
load_balance.Init(registryInst)
proxy := NewMultipleHostsReverseProxy()
log.Println("Starting httpserver at " + addr)
log.Fatal(http.ListenAndServe(addr, proxy))
}
cmd/reverseProxy/one/oneAgent.go deleted 100644 → 0
package main
import (
"bytes"
"io/ioutil"
"log"
"math/rand"
"net"
"net/http"
"net/http/httputil"
"net/url"
"regexp"
"strconv"
"strings"
"time"
)
var addr = "127.0.0.1:2001"
func main() {
rs1 := "http://127.0.0.1:2002"
url1, err1 := url.Parse(rs1)
if err1 != nil {
log.Println(err1)
}
urls := []*url.URL{url1}
proxy := NewMultipleHostsReverseProxy(urls)
log.Println("Starting httpserver at " + addr)
log.Fatal(http.ListenAndServe(addr, proxy))
}
var transport = &http.Transport{
DialContext: (&net.Dialer{
Timeout: 30 * time.Second, //连接超时
KeepAlive: 30 * time.Second, //长连接超时时间
}).DialContext,
MaxIdleConns: 100, //最大空闲连接
IdleConnTimeout: 90 * time.Second, //空闲超时时间
TLSHandshakeTimeout: 10 * time.Second, //tls握手超时时间
ExpectContinueTimeout: 1 * time.Second, //100-continue 超时时间
}
func NewMultipleHostsReverseProxy(targets []*url.URL) *httputil.ReverseProxy {
//请求协调者
director := func(req *http.Request) {
//url_rewrite
//127.0.0.1:2002/dir/abc ==> 127.0.0.1:2003/base/abc ??
//127.0.0.1:2002/dir/abc ==> 127.0.0.1:2002/abc
//127.0.0.1:2002/abc ==> 127.0.0.1:2003/base/abc
re, _ := regexp.Compile("^/dir(.*)");
req.URL.Path = re.ReplaceAllString(req.URL.Path, "$1")
//随机负载均衡
targetIndex := rand.Intn(len(targets))
target := targets[targetIndex]
targetQuery := target.RawQuery
req.URL.Scheme = target.Scheme
req.URL.Host = target.Host
// url地址重写：重写前：/aa 重写后：/base/aa
req.URL.Path = singleJoiningSlash(target.Path, req.URL.Path)
if targetQuery == "" || req.URL.RawQuery == "" {
req.URL.RawQuery = targetQuery + req.URL.RawQuery
} else {
req.URL.RawQuery = targetQuery + "&" + req.URL.RawQuery
}
if _, ok := req.Header["User-Agent"]; !ok {
req.Header.Set("User-Agent", "user-agent")
}
//只在第一代理中设置此header头
req.Header.Set("X-Real-Ip", req.RemoteAddr)
}
//更改内容
modifyFunc := func(resp *http.Response) error {
//请求以下命令：curl 'http://127.0.0.1:2002/error'
if resp.StatusCode != 200 {
//获取内容
oldPayload, err := ioutil.ReadAll(resp.Body)
if err != nil {
return err
}
//追加内容
newPayload := []byte("StatusCode error:" + string(oldPayload))
resp.Body = ioutil.NopCloser(bytes.NewBuffer(newPayload))
resp.ContentLength = int64(len(newPayload))
resp.Header.Set("Content-Length", strconv.FormatInt(int64(len(newPayload)), 10))
}
return nil
}
//错误回调 ：关闭real_server时测试，错误回调
errFunc := func(w http.ResponseWriter, r *http.Request, err error) {
http.Error(w, "ErrorHandler error:"+err.Error(), 500)
}
return &httputil.ReverseProxy{
Director: director,
Transport: transport,
ModifyResponse: modifyFunc,
ErrorHandler: errFunc}
}
func singleJoiningSlash(a, b string) string {
aslash := strings.HasSuffix(a, "/")
bslash := strings.HasPrefix(b, "/")
switch {
case aslash && bslash:
return a + b[1:]
case !aslash && !bslash:
return a + "/" + b
}
return a + b
}
cmd/reverseProxy/realServer.go deleted 100644 → 0
package main
import (
"context"
"fmt"
"ichunt-micro/proxy/load_balance"
"ichunt-micro/registry"
_ "ichunt-micro/registry/etcd"
_"github.com/imroc/req"
"io"
"io/ioutil"
"log"
"net/http"
"os"
"os/signal"
"syscall"
"time"
)
func main() {
//rs1 := &RealServer{Addr: "192.168.2.232:2003"}
//rs1.Run()
//rs2 := &RealServer{Addr: "192.168.2.232:2004"}
//rs2 := &RealServer{Addr: "192.168.1.234:2004"}
//rs2 := &RealServer{Addr: "192.168.1.237:2004"}
rs2 := &RealServer{Addr: "192.168.2.246:2004"}
rs2.Run()
//服务注册
register()
//监听关闭信号
quit := make(chan os.Signal)
signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
<-quit
}
type RealServer struct {
Addr string
}
func register(){
registryInst, err := registry.InitRegistry(context.TODO(), "etcd",
registry.WithAddrs([]string{"192.168.2.232:2379"}),
registry.WithTimeout(time.Second),
registry.WithRegistryPath("/ichuntMicroService/"),
registry.WithHeartBeat(5),
)
if err != nil {
fmt.Printf("init registry failed, err:%v", err)
return
}
load_balance.Init(registryInst)
service := &registry.Service{
Name: "comment_service",
}
service.Nodes = append(service.Nodes,
//&registry.Node{
// IP: "192.168.2.232",
// Port: 2003,
// Weight:1,
//},
&registry.Node{
IP: "192.168.2.246",
Port: 2004,
Weight:4,
},
)
registryInst.Register(context.TODO(), service)
}
func (r *RealServer) Run() {
log.Println("Starting httpserver at " + r.Addr)
mux := http.NewServeMux()
mux.HandleFunc("/test", r.HelloHandler)
mux.HandleFunc("/base/error", r.ErrorHandler)
server := &http.Server{
Addr: r.Addr,
WriteTimeout: time.Second * 3,
Handler: mux,
}
go func() {
log.Fatal(server.ListenAndServe())
}()
}
func (r *RealServer) HelloHandler(w http.ResponseWriter, req *http.Request) {
//127.0.0.1:8008/abc?sdsdsa=11
//r.Addr=127.0.0.1:8008
//req.URL.Path=/abc
//time.Sleep(time.Second)
fmt.Println("host:",req.Host)
fmt.Println("header:",req.Header)
fmt.Println("cookie:",req.Cookies())
fmt.Println(req.ParseForm())
fmt.Println("post params: ",req.PostForm)
fmt.Println("url :",req.URL)
fmt.Println("url rawpath :",req.URL.RawPath)
fmt.Println("query :",req.URL.Query())
body, err := ioutil.ReadAll(req.Body)
if err != nil {
fmt.Printf("read body err, %v\n", err)
return
}
println("json:", string(body))
upath := fmt.Sprintf("http://%s%s\n", r.Addr, req.URL.Path)
realIP := fmt.Sprintf("RemoteAddr=%s,X-Forwarded-For=%v,X-Real-Ip=%v\n", req.RemoteAddr, req.Header.Get("X-Forwarded-For"),
req.Header.Get("X-Real-Ip"))
io.WriteString(w, upath)
io.WriteString(w, realIP)
}
func (r *RealServer) ErrorHandler(w http.ResponseWriter, req *http.Request) {
upath := "error handler"
w.WriteHeader(500)
io.WriteString(w, upath)
}
\ No newline at end of file
cmd/reverseProxy/two/twoAgent.go deleted 100644 → 0
package main
import (
"bytes"
"compress/gzip"
"io/ioutil"
"log"
"math/rand"
"net"
"net/http"
"net/http/httputil"
"net/url"
"regexp"
"strconv"
"strings"
"time"
)
var addr = "127.0.0.1:2002"
func main() {
//rs1 := "http://www.baidu.com"
rs1 := "http://127.0.0.1:2003"
url1, err1 := url.Parse(rs1)
if err1 != nil {
log.Println(err1)
}
//rs2 := "http://www.baidu.com"
rs2 := "http://127.0.0.1:2004"
url2, err2 := url.Parse(rs2)
if err2 != nil {
log.Println(err2)
}
urls := []*url.URL{url1, url2}
proxy := NewMultipleHostsReverseProxy(urls)
log.Println("Starting httpserver at " + addr)
log.Fatal(http.ListenAndServe(addr, proxy))
}
var transport = &http.Transport{
DialContext: (&net.Dialer{
Timeout: 30 * time.Second, //连接超时
KeepAlive: 30 * time.Second, //长连接超时时间
}).DialContext,
MaxIdleConns: 100, //最大空闲连接
IdleConnTimeout: 90 * time.Second, //空闲超时时间
TLSHandshakeTimeout: 10 * time.Second, //tls握手超时时间
ExpectContinueTimeout: 1 * time.Second, //100-continue 超时时间
}
func NewMultipleHostsReverseProxy(targets []*url.URL) *httputil.ReverseProxy {
//请求协调者
director := func(req *http.Request) {
//url_rewrite
//127.0.0.1:2002/dir/abc ==> 127.0.0.1:2003/base/abc ??
//127.0.0.1:2002/dir/abc ==> 127.0.0.1:2002/abc
//127.0.0.1:2002/abc ==> 127.0.0.1:2003/base/abc
re, _ := regexp.Compile("^/dir(.*)");
req.URL.Path = re.ReplaceAllString(req.URL.Path, "$1")
//随机负载均衡
targetIndex := rand.Intn(len(targets))
target := targets[targetIndex]
targetQuery := target.RawQuery
req.URL.Scheme = target.Scheme
req.URL.Host = target.Host
//todo 部分章节补充1
//todo 当对域名(非内网)反向代理时需要设置此项。当作后端反向代理时不需要
req.Host = target.Host
// url地址重写：重写前：/aa 重写后：/base/aa
req.URL.Path = singleJoiningSlash(target.Path, req.URL.Path)
if targetQuery == "" || req.URL.RawQuery == "" {
req.URL.RawQuery = targetQuery + req.URL.RawQuery
} else {
req.URL.RawQuery = targetQuery + "&" + req.URL.RawQuery
}
if _, ok := req.Header["User-Agent"]; !ok {
req.Header.Set("User-Agent", "user-agent")
}
//只在第一代理中设置此header头
//req.Header.Set("X-Real-Ip", req.RemoteAddr)
}
//更改内容
modifyFunc := func(resp *http.Response) error {
//请求以下命令：curl 'http://127.0.0.1:2002/error'
//todo 部分章节功能补充2
//todo 兼容websocket
if strings.Contains(resp.Header.Get("Connection"), "Upgrade") {
return nil
}
var payload []byte
var readErr error
//todo 部分章节功能补充3
//todo 兼容gzip压缩
if strings.Contains(resp.Header.Get("Content-Encoding"), "gzip") {
gr, err := gzip.NewReader(resp.Body)
if err != nil {
return err
}
payload, readErr = ioutil.ReadAll(gr)
resp.Header.Del("Content-Encoding")
} else {
payload, readErr = ioutil.ReadAll(resp.Body)
}
if readErr != nil {
return readErr
}
//异常请求时设置StatusCode
if resp.StatusCode != 200 {
payload = []byte("StatusCode error:" + string(payload))
}
//todo 部分章节功能补充4
//todo 因为预读了数据所以内容重新回写
resp.Body = ioutil.NopCloser(bytes.NewBuffer(payload))
resp.ContentLength = int64(len(payload))
resp.Header.Set("Content-Length", strconv.FormatInt(int64(len(payload)), 10))
return nil
}
//错误回调 ：关闭real_server时测试，错误回调
errFunc := func(w http.ResponseWriter, r *http.Request, err error) {
http.Error(w, "ErrorHandler error:"+err.Error(), 500)
}
return &httputil.ReverseProxy{
Director: director,
Transport: transport,
ModifyResponse: modifyFunc,
ErrorHandler: errFunc}
}
func singleJoiningSlash(a, b string) string {
aslash := strings.HasSuffix(a, "/")
bslash := strings.HasPrefix(b, "/")
switch {
case aslash && bslash:
return a + b[1:]
case !aslash && !bslash:
return a + "/" + b
}
return a + b
}
\ No newline at end of file
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or sign in to comment