當前位置：首頁 > 编程资源 > 编程问答 >内容正文

编程问答

Cluster模式潜在问题及解决方案、Web服务综合解决方案

發布時間：2024/4/19 编程问答 41 豆豆

生活随笔收集整理的這篇文章主要介紹了 Cluster模式潜在问题及解决方案、Web服务综合解决方案小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

會不斷更新！沖沖沖！跳轉連接

https://blog.csdn.net/qq_35349982/category_10317485.html

Cluster模式潛在問題及解決方案、Web服務綜合解決方案

1.一致性的哈希算法

1.1分布式與集群

分布式和集群是不?樣的，分布式?定是集群，但是集群不?定是分布式

1.2Hash算法

1.1 順序查找法

list:List[1,5,7,6,3,4,8] // 通過循環判斷來實現 for(int element: list) { if(element == n) { 如果相等，說明n存在于數據集中 }}

1.2?分查找法

1.3 直接尋址法

定義?個數組，數組?度?于等于數據集?度，此處?度為9，數據1就存儲在下標為1的位置，3就存儲在下標為3的元素位置，，，依次類推。

這個時候，我想看下5存在與否，只需要判斷list.get(5) array[5] 是否為空，如果為空，代表5不存在于數據集，如果不為空代表5在數據集當中，通過?次查找就達到了?的，時間復雜度為O(1)。這種?式叫做“直接尋址法”：

1.4 開放尋址法 | 拉鏈法

1.3?致性Hash算法

?先有?條直線，直線開頭和結尾分別定為為1和2的32次?減1，這相當于?個地址，對于這樣?條線，彎過來構成?個圓環形成閉環，這樣的?個圓環稱為hash環。

紅節點 —服務器節點

綠人像 —請求訪問

每?臺服務器負責?段，?致性哈希算法對于節點的增減都只需重定位環空間中的??部分數據，具有較好的容錯性和可擴展性。

但是，?致性哈希算法在服務節點太少時，容易因為節點分部不均勻?造成數據傾斜問題。

例如系統中只有兩臺服務器，其環分布如下，節點2只能負責?常?的?段，?量的客戶端請求落在了節點1上，這就是數據（請求）傾斜問題

解決方案：?致性哈希算法引?了虛擬節點機制，即對每?個服務節點計算多個哈希，每個計算結果位置都放置?個此服務節點，稱為虛擬節點。

1.4手寫Hash算法

1.流程口述

聲明服務器的節點，計算hashCode然后存到SortedMap(上圈)中,(為解決數據傾斜問題，增加虛擬節點)

聲明客戶端的節點

請求到來時：那客戶端的ip的hashCode值去服務器節點中訪問，有則取，沒有則取第一個

SortedMap<Integer, String> integerStringSortedMap = hashServerMap.tailMap(clientHash);

2.代碼

import java.util.SortedMap; import java.util.TreeMap; import java.util.UUID;public class ConsistentHashWithVirtual {public static void main(String[] args) {//step1 初始化：把服務器節點IP的哈希值對應到哈希環上// 定義服務器ipString[] tomcatServers = new String[]{"123.111.0.0","123.101.3.1","111.20.35.2","123.98.26.3"};SortedMap<Integer,String> hashServerMap = new TreeMap<>();// 定義針對每個真實服務器虛擬出來幾個節點int virtaulCount = 3;for(String tomcatServer: tomcatServers) {// 求出每一個ip的hash值，對應到hash環上，存儲hash值與ip的對應關系int serverHash = Math.abs(tomcatServer.hashCode());// 存儲hash值與ip的對應關系hashServerMap.put(serverHash,tomcatServer);// 處理虛擬節點for(int i = 0; i < virtaulCount; i++) {int virtualHash = Math.abs((tomcatServer + "#" + i).hashCode());hashServerMap.put(virtualHash,"----由虛擬節點"+ i + "映射過來的請求："+ tomcatServer);}}//step2 針對客戶端IP求出hash值// 定義客戶端IPString[] clients = new String[]{"10.78.12.3","113.25.63.1","126.12.3.8"};for(String client : clients) {int clientHash = Math.abs(client.hashCode());//step3 針對客戶端,找到能夠處理當前客戶端請求的服務器（哈希環上順時針最近）// 根據客戶端ip的哈希值去找出哪一個服務器節點能夠處理（）SortedMap<Integer, String> integerStringSortedMap = hashServerMap.tailMap(clientHash);if(integerStringSortedMap.isEmpty()) {// 取哈希環上的順時針第一臺服務器Integer firstKey = hashServerMap.firstKey();System.out.println("==========>>>>客戶端：" + client + " 被路由到服務器：" + hashServerMap.get(firstKey));}else{Integer firstKey = integerStringSortedMap.firstKey();System.out.println("==========>>>>客戶端：" + client + " 被路由到服務器：" + hashServerMap.get(firstKey));}}}}

1.5nginx的負載均衡

1.ip_hash的問題

Nginx的IP_hash策略可以在客戶端ip不變的情況下，將其發出的請求始終路由到同?個?標服務器上，實現會話粘滯，避免處理session共享問題

如果沒有IP_hash策略，那么如何實現會話粘滯？

可以維護?張映射表，存儲客戶端IP或者sessionid與具體?標服務器的映射關系

<ip,tomcat1>

缺點

1）那么，在客戶端很多的情況下，映射表?常?，浪費內存空間

2）客戶端上下線，?標服務器上下線，都會導致重新維護映射表，映射表維護成本很?

2.ngx_http_upstream_consistent_hash模塊

ngx_http_upstream_consistent_hash 模塊是?個負載均衡器，使??個內部?致性hash算法來選擇合適的后端節點。

下載地址: https://github.com/replay/ngx_http_consistent_hash

配置策略

consistent_hash $remote_addr：可以根據客戶端ip映射 consistent_hash $request_uri：根據客戶端請求的uri映射 consistent_hash $args：根據客戶端攜帶的參數進?映

安裝模塊

./configure —add-module=/root/ngx_http_consistent_hash-master make make install #負載均衡策略 upstream myServer{ #每個請求按照ip的hash結果分配，每?個客戶端的請求會固定分配到同?個?標服務器處理，可以解決session問題consistent_hash $request_uri;#weight代表權重，默認每?個負載的服務器都為1，權重越?那么被分配的請求越多（?于服務器性能不均衡的場景）server 47.95.1.96:8080 weight=1;server 47.95.1.96:8081 weight=2; }proxy_pass http://myServer/;

2.時鐘同步問題

1.場景

一個電商系統，三個服務器，三個訂單，因為時鐘不同步，造成三個系統的`訂單時間`不一致

2.方案

2.1訪問授時中心

#使? ntpdate ?絡時間同步命令ntpdate -u ntp.api.bz #從?個時間服務器同步時間

2.2一個節點訪問，其他節點同步

?個服務器節點可以訪問互聯?或者所有節點都不能夠訪問互聯?

一個節點訪問，其他節點同步

##修改/etc/ntp.conf?件

#1.先讓一個節點同步授時中心 ntpdate -u ntp.api.bz #1、如果有 restrict default ignore，注釋掉它 #2、添加如下??內容restrict 172.17.0.0 mask 255.255.255.0 nomodify notrap # 放開局域?同步功能,172.17.0.0是你的局域??段server 127.127.1.0 # local clockfudge 127.127.1.0 stratum 10 #選擇一個主節點 #3、重啟?效并配置ntpd服務開機?啟動service ntpd restartchkconfig ntpd on##其他節點同步ntpdate 172.17.0.17

3.分布式ID解決?案

3.1 UUID

UUID 是指Universally Unique Identififier，翻譯為中?是通?唯?識別碼

public class MyTest {public static void main(String[] args) {System.out.println(java.util.UUID.randomUUID().toString());} }

3.2 獨?數據庫的?增ID

場景：分表后兩張表公用一個獨立表的ID

創建一張表

-- ---------------------------- -- Table structure for DISTRIBUTE_ID -- ---------------------------- DROP TABLE IF EXISTS `DISTRIBUTE_ID`; CREATE TABLE `DISTRIBUTE_ID` (`id` bigint(32) NOT NULL AUTO_INCREMENT COMMENT '主鍵',`createtime` datetime DEFAULT NULL,PRIMARY KEY (`id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8; insert into DISTRIBUTE_ID(createtime) values(NOW()); select LAST_INSERT_ID()；

缺點:

需要代碼連接到數據庫才能獲取到id，性能?法保障，

mysql數據庫實例掛掉了，那么就?法獲取分布式id了。

3.3 SnowFlake 雪花算法

3.4 借助Redis的Incr命令獲取全局唯?ID

安裝redis后

修改配置文件 redis.conf

啟動

#bind 1270.0.1注釋掉 protected-mode no #設置為no <dependency><groupId>redis.clients</groupId><artifactId>jedis</artifactId><version>2.9.0</version> </dependency> Jedis jedis = new Jedis("127.0.0.1",6379); try {long id = jedis.incr("id");//定義的KEY 是IDSystem.out.println("從redis中獲取的分布式id為：" + id); } finally {if (null != jedis) {jedis.close();} }

源碼：

public Long incr(String key) {this.checkIsInMultiOrPipeline();//獲取鏈接this.client.incr(key); //獲取keyreturn this.client.getIntegerReply();} //======================================= //讀取 public long readLongCrLf() {byte[] buf = this.buf;this.ensureFill();boolean isNeg = buf[this.count] == 45;if (isNeg) {++this.count;}long value = 0L;while(true) {this.ensureFill();int b = buf[this.count++];if (b == 13) {this.ensureFill();if (buf[this.count++] != 10) {throw new JedisConnectionException("Unexpected character!");} else {return isNeg ? -value : value;}}value = value * 10L + (long)b - 48L;}}

4.分布式任務調度

4.1定時任務場景

訂單審核，出庫
訂單支付退款
禮券同步
物流信息推送
短信推送
日志監控

4.2什么是分布式任務調度

在分布式集群環境下的調度任務（同一個定時任務部署多分，只執行一個定時任務）

分布式調度->定時任務的分布式->定時任務拆分（把一個大任務拆分成切分，同時執行）

4.3定時任務與消息隊列的區別

共同點
異步處理

比如注冊，下單時間
應用解耦

不管定時任務作業還是MQ 都可以實現應用解耦
流量削峰

任何作業和MQ都可以用來抗流量
不同點
定時任務作業時時間驅動，MQ是時間驅動
時間驅動是不可替代的，比如說金融系統是每日的利息結算，不是來一條做一條，還是批量時間去處理
定時任務作業更傾向于批處理，MQ傾向于逐條處理

4.4 Quartz定時

<dependency><groupId>org.quartz-scheduler</groupId><artifactId>quartz</artifactId><version>2.3.2</version></dependency> public class QuartzMan {// 1、創建任務調度器（好比公交調度站）public static Scheduler createScheduler() throws SchedulerException {SchedulerFactory schedulerFactory = new StdSchedulerFactory();Scheduler scheduler = schedulerFactory.getScheduler();return scheduler;}// 2、創建一個任務（好比某一個公交車的出行）public static JobDetail createJob() {JobBuilder jobBuilder = JobBuilder.newJob(DemoJob.class); // TODO 自定義任務類jobBuilder.withIdentity("jobName","myJob");JobDetail jobDetail = jobBuilder.build();return jobDetail;}/*** 3、創建作業任務時間觸發器（類似于公交車出車時間表）* cron表達式由七個位置組成，空格分隔* 1、Seconds（秒） 0~59* 2、Minutes（分） 0~59* 3、Hours（小時） 0~23* 4、Day of Month（天）1~31,注意有的月份不足31天* 5、Month（月） 0~11,或者 JAN,FEB,MAR,APR,MAY,JUN,JUL,AUG,SEP,OCT,NOV,DEC* 6、Day of Week(周) 1~7,1=SUN或者 SUN,MON,TUE,WEB,THU,FRI,SAT* 7、Year（年）1970~2099 可選項*示例：* 0 0 11 * * ? 每天的11點觸發執行一次* 0 30 10 1 * ? 每月1號上午10點半觸發執行一次*/public static Trigger createTrigger() {// 創建時間觸發器CronTrigger cronTrigger = TriggerBuilder.newTrigger().withIdentity("triggerName","myTrigger").startNow().withSchedule(CronScheduleBuilder.cronSchedule("*/2 * * * * ?")).build();return cronTrigger;}/*** main函數中開啟定時任務* @param args*/public static void main(String[] args) throws SchedulerException {// 1、創建任務調度器（好比公交調度站）Scheduler scheduler = QuartzMan.createScheduler();// 2、創建一個任務（好比某一個公交車的出行）JobDetail job = QuartzMan.createJob();// 3、創建任務的時間觸發器（好比這個公交車的出行時間表）Trigger trigger = QuartzMan.createTrigger();// 4、使用任務調度器根據時間觸發器執行我們的任務scheduler.scheduleJob(job,trigger);scheduler.start();} }

4.5 Elastic-job

github地址：https://github.com/elasticjob

1.主要功能

分布式環境下，能夠避免同一個任務多實例重復執行
彈性擴容縮容，當集群中增加某一個實例，它能被選舉且執行任務，減少一個實例，任務自動分配給其他實例
任務失敗時，轉移給其他實例
自動記錄錯過執行的作業
支持并行調度，支持任務分片
作業分片后，能后保證同一分片在分布式環境中僅執行一個

2.代碼執行實例

分片 ====>shardingItemParameters(“0=bachelor,1=master,2=doctor”)
把一個任務分成多個task,每一個task交給一個具體的一個機器實例去處理
Strategy策略定義這些分片項怎么去分配到機器，默認是平均分，分片和作業是通過一個注冊中心協調的
擴容

分?項也是?個JOB配置，修改配置，重新分?，在下?次定時運?之前會重新調?分?算法，那么這個分?算法的結果就是：哪臺機器運?哪?個??，這個結果存儲到zk中的，主節點會把分?給分好放到注冊中?去，然后執?節點從注冊中?獲取信息(執?節點在定時任務開啟的時候獲取相應的分?)

高可用，其他節點全部掛點了，剩下一個節點，則所有的節點都指向這個節點

<dependency><groupId>com.dangdang</groupId><artifactId>elastic-job-lite-core</artifactId><version>2.1.5</version></dependency> //任務調度類 public static void main(String[] args) {// 配置分布式協調服務（注冊中心）ZookeeperZookeeperConfiguration zookeeperConfiguration = new ZookeeperConfiguration("localhost:2181","data-archive-job");CoordinatorRegistryCenter coordinatorRegistryCenter = new ZookeeperRegistryCenter(zookeeperConfiguration);//初始化coordinatorRegistryCenter.init();// 配置任務（時間事件、定時任務業務邏輯、調度器）JobCoreConfiguration jobCoreConfiguration = JobCoreConfiguration.newBuilder("archive-job", "*/2 * * * * ?", 3).shardingItemParameters("0=bachelor,1=master,2=doctor").build();SimpleJobConfiguration simpleJobConfiguration = new SimpleJobConfiguration(jobCoreConfiguration,ArchivieJob.class.getName());//任務初始化//1.配置任分布式協調中心//2.配置任務，執行的任務類，定時時間JobScheduler jobScheduler = new JobScheduler(coordinatorRegistryCenter, LiteJobConfiguration.newBuilder(simpleJobConfiguration).overwrite(true).build());jobScheduler.init();} /*** ElasticJobLite定時任務業務邏輯處理類*/ public class ArchivieJob implements SimpleJob {/*** 需求：resume表中未歸檔的數據歸檔到resume_bak表中，每次歸檔1條記錄* execute方法中寫我們的業務邏輯（execute方法每次定時任務執行都會執行一次）* @param shardingContext*/@Overridepublic void execute(ShardingContext shardingContext) {int shardingItem = shardingContext.getShardingItem();System.out.println("=====>>>>當前分片：" + shardingItem);// 獲取分片參數String shardingParameter = shardingContext.getShardingParameter(); // 0=bachelor,1=master,2=doctor// 1 從resume表中查詢出1條記錄（未歸檔）String selectSql = "select * from resume where state='未歸檔' and education='"+ shardingParameter +"' limit 1";List<Map<String, Object>> list = JdbcUtil.executeQuery(selectSql);if(list == null || list.size() ==0 ) {System.out.println("數據已經處理完畢！！！！！！");return;}// 2 "未歸檔"更改為"已歸檔"Map<String, Object> stringObjectMap = list.get(0);long id = (long) stringObjectMap.get("id");String name = (String) stringObjectMap.get("name");String education = (String) stringObjectMap.get("education");System.out.println("=======>>>>id：" + id + " name：" + name + " education：" + education);String updateSql = "update resume set state='已歸檔' where id=?";JdbcUtil.executeUpdate(updateSql,id);// 3 歸檔這條記錄，把這條記錄插入到resume_bak表String insertSql = "insert into resume_bak select * from resume where id=?";JdbcUtil.executeUpdate(insertSql,id);} }

分布式的理解

5.Session的共享

5.1 Session問題原因分析

Http是無狀態協議

兩種用于保持Http狀態的技術，那就是Cookie和Session

JSESSIONID是一個Cookie，Servlet容器（tomcat，jetty）用來記錄用戶session

5.2 Session一致性的解決方案

nginx的ip_hash策略

同?個客戶端IP的請求都會被路由到同?個?標服務器，也叫做會話粘滯

優點：

配置簡單，不?侵應?，不需要額外修改代碼

缺點：

服務器重啟Session丟失

存在單點負載?的?險

單點故障問題

Session復制（不推薦）多個tomcat之間通過修改配置?件，達到Session之間的復制

Session共享(交代redis)

優點：

能適應各種負載均衡策略
服務器重啟或者當即不會造成Session丟失
擴展能力強
適合大集群數量使用

缺點

對應用有侵入性

5.3代碼實現(Session共享)

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId></dependency><dependency><groupId>org.springframework.session</groupId><artifactId>spring-session-data-redis</artifactId></dependency> spring.redis.database=0 spring.redis.host=127.0.0.1 spring.redis.port=6379 @SpringBootApplication @EnableCaching @EnableRedisHttpSession //加這個注解 public class LoginApplication extends SpringBootServletInitializer {@Overrideprotected SpringApplicationBuilder configure(SpringApplicationBuilder builder) {return builder.sources(LoginApplication.class);}public static void main(String[] args) {SpringApplication.run(LoginApplication.class, args);}}

5.4源碼刨析

引入了RedisHttpSessionConfiguration類

繼承了SpringHttpSessionConfiguration類

SessionRepositoryFilter類

//1.引入了RedisHttpSessionConfiguration類 @Import({RedisHttpSessionConfiguration.class}) public @interface EnableRedisHttpSession { } //=========================== //2.繼承了SpringHttpSessionConfiguration類 public class RedisHttpSessionConfiguration extends SpringHttpSessionConfiguration {} //=========================== //3.SessionRepositoryFilter類 public class SpringHttpSessionConfiguration{@Beanpublic <S extends Session> SessionRepositoryFilter<? extends Session> springSessionRepositoryFilter(SessionRepository<S> sessionRepository) {SessionRepositoryFilter<S> sessionRepositoryFilter = new SessionRepositoryFilter(sessionRepository);sessionRepositoryFilter.setHttpSessionIdResolver(this.httpSessionIdResolver);return sessionRepositoryFilter;} } //protected void doFilterInternal(HttpServletRequest request, HttpServletResponse response, FilterChain filterChain) throws ServletException, IOException {request.setAttribute(SESSION_REPOSITORY_ATTR, this.sessionRepository);SessionRepositoryFilter<S>.SessionRepositoryRequestWrapper wrappedRequest = new SessionRepositoryFilter.SessionRepositoryRequestWrapper(request, response);SessionRepositoryFilter.SessionRepositoryResponseWrapper wrappedResponse = new SessionRepositoryFilter.SessionRepositoryResponseWrapper(wrappedRequest, response);try {filterChain.doFilter(wrappedRequest, wrappedResponse);} finally {wrappedRequest.commitSession();//redis提交}} //redis提交private void commitSession() {SessionRepositoryFilter<S>.SessionRepositoryRequestWrapper.HttpSessionWrapper wrappedSession = this.getCurrentSession();if (wrappedSession == null) {if (this.isInvalidateClientSession()) {SessionRepositoryFilter.this.httpSessionIdResolver.expireSession(this, this.response);}} else {S session = wrappedSession.getSession();this.clearRequestedSessionCache();SessionRepositoryFilter.this.sessionRepository.save(session);String sessionId = session.getId();if (!this.isRequestedSessionIdValid() || !sessionId.equals(this.getRequestedSessionId())) {SessionRepositoryFilter.this.httpSessionIdResolver.setSessionId(this, this.response, sessionId);}}}

附錄

1.wrapper的概念

實現Wrapper的類需要:

實現某個接口

依賴該接口的子類, 并通過接口方法訪問子類對象的方法

public class MyArrays {public List<Point> asList(Point[] points) {List<Point> list = new ArrayList<Point>();for (int i = 0; i < points.length; i++) {list.add(points[i]);}return new ArrayListWrapper(list);} }class ArrayListWrapper implements List<Point> {private List<Point> list;ArrayListWrapper(List<Point> list) {this.list = list;}@Overridepublic int size() {return list.size();}@Overridepublic boolean add(Point e) {throw new UnsupportedOperationException("...");}@Overridepublic boolean remove(Object o) {throw new UnsupportedOperationException("...");} …………………………}

單詞

Cluster 群，集群

strict 嚴格

stratum 層

ensure 保證
Coordinator協調，班主任

elastic 彈性的

總結

以上是生活随笔為你收集整理的Cluster模式潜在问题及解决方案、Web服务综合解决方案的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。