微服务消息推送的Consul服务治理发现

1:新建信息服务项目 MsgService 右键新建项目选择下图所示的选项

下一步然后选择ASP.NET Core2.1的Api

2: 在MsgService 项目的控制器中新建Api控制器 EmailController与SMSController 控制器中的代码如下:

EmailController:

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Microsoft.AspNetCore.Http;
using Microsoft.AspNetCore.Mvc;namespace MsgService.Controllers
{[Route("api/[controller]")][ApiController]public class EmailController : ControllerBase{[HttpPost(nameof(Send_QQ))]public void Send_QQ(SendEmailRequest model){Console.WriteLine($"通过QQ邮件接口向{model.Email}发送邮件，标题{model.Title}，内容：{ model.Body}");}[HttpPost(nameof(Send_163))]public void Send_163(SendEmailRequest model){Console.WriteLine($"通过网易邮件接口向{model.Email}发送邮件，标题{model.Title}，内容：{ model.Body} ");}[HttpPost(nameof(Send_Sohu))]public void Send_Sohu(SendEmailRequest model){Console.WriteLine($"通过Sohu邮件接口向{model.Email}发送邮件，标题{model.Title}，内容：{ model.Body}");}}public class SendEmailRequest{public string Email { get; set; }public string Title { get; set; }public string Body { get; set; }}
}

SMSController :

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Microsoft.AspNetCore.Http;
using Microsoft.AspNetCore.Mvc;namespace MsgService.Controllers
{[Route("api/[controller]")][ApiController]public class SMSController : ControllerBase{//发请求，报文体为{phoneNum:"110",msg:"aaaaaaaaaaaaa"}，[HttpPost(nameof(Send_MI))]public void Send_MI(dynamic model){Console.WriteLine($"通过小米短信接口向{model.phoneNum}发送短信{model.msg}");}[HttpPost(nameof(Send_LX))]public void Send_LX(SendSMSRequest model){Console.WriteLine($"通过联想短信接口向{model.PhoneNum}发送短信{model.Msg}");}[HttpPost(nameof(Send_HW))]public void Send_HW(SendSMSRequest model){Console.WriteLine($"通过华为短信接口向{model.PhoneNum}发送短信{model.Msg}");}}public class SendSMSRequest{public string PhoneNum { get; set; }public string Msg { get; set; }}
}

3:新建产品信息服务项目 ProductService 创建该项目方法与上面的MsgService一样

然后创建控制器ProductController.cs 如下：

using System;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Threading.Tasks;
using Microsoft.AspNetCore.Http;
using Microsoft.AspNetCore.Mvc;namespace ProductService.Controllers
{[Route("api/[controller]")][ApiController]public class ProductController : ControllerBase{//这显然是为了demo，这样放到内存中不能“集群”private static List<Product> products = new List<Product>();static ProductController(){products.Add(new Product{Id = 1,Name = "T430笔记本",Price = 8888,Description = "CPU i7标压版，1T硬盘"});products.Add(new Product{Id = 2,Name = "华为Mate10",Price = 3888,Description = "大猩猩屏幕，多点触摸"});products.Add(new Product{Id = 3,Name = "天梭手表",Price = 9888,Description = "瑞士 经典款，可好了"});}[HttpGet]public IEnumerable<Product> Get(){return products;}[HttpGet("{id}")]public Product Get(int id){var product = products.SingleOrDefault(p => p.Id == id);if (product == null){Response.StatusCode = 404;}return product;}[HttpPost]public void Add(Product model){if (products.Any(p => p.Id == model.Id)){Response.StatusCode = (int)HttpStatusCode.Conflict;//通过状态码而非响应体报错，是restful风格return;}products.Add(model);}[HttpDelete("{id}")]public void Delete(int id){var product = products.SingleOrDefault(p => p.Id == id);if (product != null){products.Remove(product);}}}public class Product{public long Id { get; set; }public string Name { get; set; }public double Price { get; set; }public string Description { get; set; }}
}

4：consul 服务器安装

consul 下载地址 https://www.consul.io/downloads.html

因为我的电脑是windows64位的，因为仅仅是测试，所有就下了一个 windows版本的Consul

下载下来后，在运行中输入cmd 命令，然后在cmd中输入命令，执行cd命令切换到consul文件所在的盘符及文件夹

然后执行 consul.exe agent -dev

这是开发环境测试，生产环境要建集群，要至少一台 Server，多台 Agent。开发环境中 consul 重启后数据就会丢失。

consul 的监控页面 http://127.0.0.1:8500/

consult 主要做三件事：提供服务到 ip 地址的注册；提供服务到 ip 地址列表的查询；对提供服务方的健康检（HealthCheck）；

5：.Net Core 连接 consul

在MsgService与ProductService两个项目中都添加健康检查控制器 HealthController 如下：

 public class HealthController : ControllerBase{[HttpGet]public IActionResult Get(){return Ok("ok");}}

同时两个项目中皆添加 Nuget包命令：Install-Package Consul

6:服务注册 Consul 及注销

在MsgService与ProductService两个项目的Startup.cs的Configure方法中修改代码如下：

// This method gets called by the runtime. Use this method to configure the HTTP request pipeline.public void Configure(IApplicationBuilder app, IHostingEnvironment env, IApplicationLifetime applicationLifetime)//添加一个参数IApplicationLifetime applicationLifetime{if (env.IsDevelopment()){app.UseDeveloperExceptionPage();}else{app.UseHsts();}app.UseHttpsRedirection();app.UseMvc();#region 添加部分//获取IDstring ip = Configuration["ip"];//获取端口号int port = Convert.ToInt32(Configuration["port"]);//定义服务名称string serviceName = "MsgService";//服务idstring serviceId = serviceName + Guid.NewGuid();//注册到Consulusing (var client = new ConsulClient(ConsulConfig)){//注册服务到 Consulclient.Agent.ServiceRegister(new AgentServiceRegistration(){ID = serviceId,//服务编号，不能重复，用 Guid 最简单Name = serviceName,//服务的名字Address = ip,//服务提供者的能被消费者访问的 ip 地址(可以被其他应用访问的地址，本地测试可以用 127.0.0.1，机房环境中一定要写自己的内网 ip 地址)Port = port,// 服务提供者的能被消费者访问的端口Check = new AgentServiceCheck{DeregisterCriticalServiceAfter = TimeSpan.FromSeconds(5),//服务停止多久后反注册(注销)Interval = TimeSpan.FromSeconds(10),//健康检查时间间隔，或者称为心跳间隔HTTP = $"https://{ip}:{port}/api/health",//健康检查地址Timeout = TimeSpan.FromSeconds(5)}}).Wait();//Consult 客户端的所有方法几乎都是异步方法，但是都没按照规范加上Async 后缀，所以容易误导。记得调用后要 Wait()或者 await}//程序正常退出的时候从 Consul 注销服务//要通过方法参数注入 IApplicationLifetimeapplicationLifetime.ApplicationStopped.Register(() => {using (var client = new ConsulClient(ConsulConfig)){client.Agent.ServiceDeregister(serviceId).Wait();}});}/// <summary>/// Consul的配置/// </summary>/// <param name="c"></param>private void ConsulConfig(ConsulClientConfiguration c){c.Address = new Uri("http://127.0.0.1:8500");c.Datacenter = "dc1";}#endregion

同时在两个项目中的appsettings.json配置两个节点

 "ip": "localhost","port": "44343",
//port中的44343值在每个项目中都不一样 在每个项目中的Properties下的launchSettings.json中查看port的值 如下图所示：

然后可通过命令调试实例也可通过项目直接启动新实例

cmd启动新实例的命令：

dotnet WebApplication4.dll --ip 127.0.0.1 --port 5001
//WebApplication4.dll 表示项目bin目录下的服务的dll
//ip的值表示本机
//port的值在Properties的launchSettings.json中获取

二：编写服务消费者

1：新建一个RestTemplateCore类库，nuget 安装：Consul、Newtonsoft.Json

在RestTemplateCore中添加三个类RestResponse.cs，RestResponseWithBody.cs，RestTemplate.cs

RestResponse.cs类代码如下：

/// <summary>/// Rest响应结果/// </summary>public class RestResponse{/// <summary>/// 响应状态码/// </summary>public HttpStatusCode StatusCode { get; set; }/// <summary>/// 响应的报文头/// </summary>public HttpResponseHeaders Headers { get; set; }}

RestResponseWithBody.cs

/// <summary>/// 带响应报文的Rest响应结果，而且json报文会被自动反序列化/// </summary>/// <typeparam name="T"></typeparam>public class RestResponseWithBody<T> : RestResponse{/// <summary>/// 响应报文体json反序列化的内容/// </summary>public T Body { get; set; }}

RestTemplate.cs

/// <summary>/// 会自动到 Consul 中解析服务的 Rest 客户端，能把"http://ProductService/api/Product/"这样的虚拟地址/// 按照客户端负载均衡算法解析为 http://192.168.1.10:8080/api/Product/这样的真实地址/// </summary>public class RestTemplate{public String ConsulServerUrl { get; set; } = "http://127.0.0.1:8500";private HttpClient httpClient;public RestTemplate(HttpClient httpClient){this.httpClient = httpClient;}/// <summary>/// 获取服务的第一个实现地址/// </summary>/// <param name="consulClient"></param>/// <param name="serviceName"></param>/// <returns></returns>private async Task<String> ResolveRootUrlAsync(String serviceName){//Consul包using (var consulClient = new ConsulClient(c => c.Address = newUri(ConsulServerUrl))){var services = (await consulClient.Agent.Services()).Response.Values.Where(s => s.Service.Equals(serviceName,StringComparison.OrdinalIgnoreCase));if (!services.Any()){throw new ArgumentException($"找不到服务{serviceName }的任何实例");}else{//根据当前时钟毫秒数对可用服务个数取模，取出一台机器使用var service = services.ElementAt(Environment.TickCount %services.Count());return $"{service.Address}:{service.Port}";}}}//把 http://apiservice1/api/values 转换为 http://192.168.1.1:5000/api/valuesprivate async Task<String> ResolveUrlAsync(String url){Uri uri = new Uri(url);String serviceName = uri.Host;//apiservice1String realRootUrl = await ResolveRootUrlAsync(serviceName);// 查 询 出 来apiservice1 对应的服务器地址 192.168.1.1:5000//uri.Scheme=http,realRootUrl =192.168.1.1:5000,PathAndQuery=/api/valuesreturn uri.Scheme + "://" + realRootUrl + uri.PathAndQuery;}/// <summary>/// 发出 Get 请求/// </summary>/// <typeparam name="T">响应报文反序列类型</typeparam>/// <param name="url">请求路径</param>/// <param name="requestHeaders">请求额外的报文头信息</param>/// <returns></returns>public async Task<RestResponseWithBody<T>> GetForEntityAsync<T>(String url,HttpRequestHeaders requestHeaders = null){using (HttpRequestMessage requestMsg = new HttpRequestMessage()){if (requestHeaders != null){foreach (var header in requestHeaders){requestMsg.Headers.Add(header.Key, header.Value);}}requestMsg.Method = System.Net.Http.HttpMethod.Get;//http://apiservice1/api/values 转换为 http://192.168.1.1:5000/api/valuesrequestMsg.RequestUri = new Uri(await ResolveUrlAsync(url));RestResponseWithBody<T> respEntity = awaitSendForEntityAsync<T>(requestMsg);return respEntity;}}/// <summary>/// 发出 Post 请求/// </summary>/// <typeparam name="T">响应报文反序列类型</typeparam>/// <param name="url">请求路径</param>/// <param name="body">请求数据，将会被 json 序列化后放到请求报文体中</param>/// <param name="requestHeaders">请求额外的报文头信息</param>/// <returns></returns>public async Task<RestResponseWithBody<T>> PostForEntityAsync<T>(String url, objectbody = null, HttpRequestHeaders requestHeaders = null){using (HttpRequestMessage requestMsg = new HttpRequestMessage()){if (requestHeaders != null){foreach (var header in requestHeaders){requestMsg.Headers.Add(header.Key, header.Value);}}requestMsg.Method = System.Net.Http.HttpMethod.Post;//http://apiservice1/api/values 转换为 http://192.168.1.1:5000/api/valuesrequestMsg.RequestUri = new Uri(await ResolveUrlAsync(url));requestMsg.Content = newStringContent(JsonConvert.SerializeObject(body));requestMsg.Content.Headers.ContentType = newMediaTypeHeaderValue("application/json");RestResponseWithBody<T> respEntity = awaitSendForEntityAsync<T>(requestMsg);return respEntity;}}/// <summary>/// 发出 Post 请求/// </summary>/// <param name="url">请求路径</param>/// <param name="requestHeaders">请求额外的报文头信息</param>/// <returns></returns>public async Task<RestResponse> PostAsync(String url, object body = null,HttpRequestHeaders requestHeaders = null){using (HttpRequestMessage requestMsg = new HttpRequestMessage()){if (requestHeaders != null){foreach (var header in requestHeaders){requestMsg.Headers.Add(header.Key, header.Value);}}requestMsg.Method = System.Net.Http.HttpMethod.Post;//http://apiservice1/api/values 转换为 http://192.168.1.1:5000/api/valuesrequestMsg.RequestUri = new Uri(await ResolveUrlAsync(url));requestMsg.Content = newStringContent(JsonConvert.SerializeObject(body));requestMsg.Content.Headers.ContentType = newMediaTypeHeaderValue("application/json");var resp = await SendAsync(requestMsg);return resp;}}/// <summary>/// 发出 Put 请求/// </summary>/// <typeparam name="T">响应报文反序列类型</typeparam>/// <param name="url">请求路径</param>/// <param name="body">请求数据，将会被 json 序列化后放到请求报文体中</param>/// <param name="requestHeaders">请求额外的报文头信息</param>/// <returns></returns>public async Task<RestResponseWithBody<T>> PutForEntityAsync<T>(String url, objectbody = null, HttpRequestHeaders requestHeaders = null){using (HttpRequestMessage requestMsg = new HttpRequestMessage()){if (requestHeaders != null){foreach (var header in requestHeaders){requestMsg.Headers.Add(header.Key, header.Value);}}requestMsg.Method = System.Net.Http.HttpMethod.Put;//http://apiservice1/api/values 转换为 http://192.168.1.1:5000/api/valuesrequestMsg.RequestUri = new Uri(await ResolveUrlAsync(url));requestMsg.Content = newStringContent(JsonConvert.SerializeObject(body));requestMsg.Content.Headers.ContentType = newMediaTypeHeaderValue("application/json");RestResponseWithBody<T> respEntity = awaitSendForEntityAsync<T>(requestMsg);return respEntity;}}/// <summary>/// 发出 Put 请求/// </summary>/// <param name="url">请求路径</param>/// <param name="body">请求数据，将会被 json 序列化后放到请求报文体中</param>/// <param name="requestHeaders">请求额外的报文头信息</param>/// <returns></returns>public async Task<RestResponse> PutAsync(String url, object body = null,HttpRequestHeaders requestHeaders = null){using (HttpRequestMessage requestMsg = new HttpRequestMessage()){if (requestHeaders != null){foreach (var header in requestHeaders){requestMsg.Headers.Add(header.Key, header.Value);}}requestMsg.Method = System.Net.Http.HttpMethod.Put;//http://apiservice1/api/values 转换为 http://192.168.1.1:5000/api/valuesrequestMsg.RequestUri = new Uri(await ResolveUrlAsync(url));requestMsg.Content = newStringContent(JsonConvert.SerializeObject(body));requestMsg.Content.Headers.ContentType = newMediaTypeHeaderValue("application/json");var resp = await SendAsync(requestMsg);return resp;}}/// <summary>/// 发出 Delete 请求/// </summary>/// <typeparam name="T">响应报文反序列类型</typeparam>/// <param name="url">请求路径</param>/// <param name="requestHeaders">请求额外的报文头信息</param>/// <returns></returns>public async Task<RestResponseWithBody<T>> DeleteForEntityAsync<T>(String url,HttpRequestHeaders requestHeaders = null){using (HttpRequestMessage requestMsg = new HttpRequestMessage()){if (requestHeaders != null){foreach (var header in requestHeaders){requestMsg.Headers.Add(header.Key, header.Value);}}requestMsg.Method = System.Net.Http.HttpMethod.Delete;//http://apiservice1/api/values 转换为 http://192.168.1.1:5000/api/valuesrequestMsg.RequestUri = new Uri(await ResolveUrlAsync(url));RestResponseWithBody<T> respEntity = awaitSendForEntityAsync<T>(requestMsg);return respEntity;}}/// <summary>/// 发出 Delete 请求/// </summary>/// <param name="url">请求路径</param>/// <param name="requestHeaders">请求额外的报文头信息</param>/// <returns></returns>public async Task<RestResponse> DeleteAsync(String url, HttpRequestHeadersrequestHeaders = null){using (HttpRequestMessage requestMsg = new HttpRequestMessage()){if (requestHeaders != null){foreach (var header in requestHeaders){requestMsg.Headers.Add(header.Key, header.Value);}}requestMsg.Method = System.Net.Http.HttpMethod.Delete;//http://apiservice1/api/values 转换为 http://192.168.1.1:5000/api/valuesrequestMsg.RequestUri = new Uri(await ResolveUrlAsync(url));var resp = await SendAsync(requestMsg);return resp;}}/// <summary>/// 发出 Http 请求/// </summary>/// <typeparam name="T">响应报文反序列类型</typeparam>/// <param name="requestMsg">请求数据</param>/// <returns></returns>public async Task<RestResponseWithBody<T>>SendForEntityAsync<T>(HttpRequestMessage requestMsg){var result = await httpClient.SendAsync(requestMsg);RestResponseWithBody<T> respEntity = new RestResponseWithBody<T>();respEntity.StatusCode = result.StatusCode;respEntity.Headers = respEntity.Headers;String bodyStr = await result.Content.ReadAsStringAsync();if (!string.IsNullOrWhiteSpace(bodyStr)){respEntity.Body = JsonConvert.DeserializeObject<T>(bodyStr);}return respEntity;}/// <summary>/// 发出 Http 请求/// </summary>/// <param name="requestMsg">请求数据</param>/// <returns></returns>public async Task<RestResponse> SendAsync(HttpRequestMessage requestMsg){var result = await httpClient.SendAsync(requestMsg);RestResponse response = new RestResponse();response.StatusCode = result.StatusCode;response.Headers = result.Headers;return response;}}

2:新建控制台项目consultest1进行测试消费

在程序入口Program.cs中添加方法:

class Program{static void Main(string[] args){using (HttpClient httpClient = new HttpClient()){RestTemplate rest = new RestTemplate(httpClient);Console.WriteLine("---查询数据---------");var ret1 =rest.GetForEntityAsync<Product[]>("https://ProductService/api/Product/").Result;Console.WriteLine(ret1.StatusCode);if (ret1.StatusCode == System.Net.HttpStatusCode.OK){foreach (var p in ret1.Body){Console.WriteLine($"id={p.Id},name={p.Name}");}}Console.WriteLine("---新增数据---------");Product newP = new Product();newP.Id = 888;newP.Name = "辛增";newP.Price = 88.8;var ret = rest.PostAsync("https://ProductService/api/Product/", newP).Result;Console.WriteLine(ret.StatusCode);}Console.ReadKey();}class Product{public long Id { get; set; }public string Name { get; set; }public double Price { get; set; }public string Description { get; set; }}}

测试：先启动上述MsgService，ProductService两个项目

然后在进入监控页面查看：http://127.0.0.1:8500/

最后启动consultest1控制台项目看效果就行了

微服务消息推送的Consul服务治理发现相关推荐

APNS提供了两项基本的服务：消息推送和反馈服务
推送通知,也被叫做远程通知,是在iOS 3.0以后被引入的功能.是当程序没有启动或不在前台运行时,告诉用户有新消息的一种途径,是从外部服务器发送到应用程序上的.一般说来,当要显示消息或下载数据的时候, ...
java 消息推送_hanbo-push分布式消息推送、IM服务
系统概览 app接入除了接入restApi(push-admin)之外,还需要兼容connector(push-admin的client,用于和push-server通信)的通信协议. 基于proto ...
dwr消息推送和tomcat集群
网友的提问: 项目中用到了dwr消息推送.而服务端是通过一个http请求后触发dwr中的推送方法.而单个tomcat中.服务器发送的http请求和用户都在一个tomcat服务器中.这样就能精准推送到 ...
android热门消息推送横向测评！[转]
关于这个话题,已经不是什么新鲜事了.对于大多数中小型公司一般都是选择第三方的服务来实现.但是现在已经有很多提供推送服务的公司和产品,如何选择一个适合自己项目的服务呢?它们之间都有什么差别?在此为大家做 ...
Java之消息推送浅入浅出
在日常开发中,消息推送是非常典型的业务需求,下面对消息推送简单的分析一下. 消息推送通常指网站的运营人员通过某种工具对用户当前网页或移动设备APP进行的主动消息推送.主要分为web端消息推送和移动端消 ...
Android 消息推送框架详解
消息推送的概念消息推送,是指绕过手机运营商,通过TCP/IP网络传输的方式,向应用程序发送数据,这些数据包括简单的文本,图片,或者其他多媒体数据. 与手机运营商发送短信的方式相比,消息推送普及性和可 ...
android消息推送标准,如何提升Android消息推送的到达率？
消息推送时下已经是app日常运营的重要工具,app的消息推送与一个app的日活息息相关,好的消息推送能够有效地增强用户黏性,然而一切好的消息推送都是建立在能够到达用户终端为前提的,所以消息推送的到达率 ...
android热门消息推送横向测评！
关于这个话题,已经不是什么新鲜事了.对于大多数中小型公司一般都是选择第三方的服务来实现.但是现在已经有很多提供推送服务的公司和产品,如何选择一个适合自己项目的服务呢?它们之间都有什么差别?在此为大家做 ...
HMS Core地理围栏能力助你实现指定范围人群的精准消息推送
精准推送是移动端产品留存阶段的主要运营手段,精准推送常常会与用户画像紧密结合,针对用户的喜好.画像,采用不同策略,但基于用户所属区域推送消息却很难实现.目前市面上大多数第三方消息推送服务商,在系统未深 ...

微服务消息推送的Consul服务治理发现

二：编写服务消费者

微服务消息推送的Consul服务治理发现相关推荐

最新文章

热门文章