分布式id号段获取

im-logic-api/im-segment-id-api/src/main/java/net/sopod/soim/logic/segmentid/api/model/Segment.java

@@ -1,4 +1,4 @@
-package net.sopod.soim.logic.segmentid.model;
+package net.sopod.soim.logic.segmentid.api.model;
 
 /**
  * Segment

im-logic-api/im-segment-id-api/src/main/java/net/sopod/soim/logic/segmentid/api/model/dto/NextSegmentParam.java

@@ -1,4 +1,4 @@
-package net.sopod.soim.logic.segmentid.model.dto;
+package net.sopod.soim.logic.segmentid.api.model.dto;
 
 import lombok.Data;
 

im-logic-api/im-segment-id-api/src/main/java/net/sopod/soim/logic/segmentid/api/model/dto/SegmentDTO.java

@@ -1,8 +1,10 @@
-package net.sopod.soim.logic.segmentid.model.dto;
+package net.sopod.soim.logic.segmentid.api.model.dto;
 
 import lombok.Data;
 import lombok.experimental.Accessors;
 
+import java.io.Serializable;
+
 /**
  * SegmentDTO
  *
@@ -11,7 +13,9 @@ import lombok.experimental.Accessors;
  */
 @Data
 @Accessors(chain = true)
-public class SegmentDTO {
+public class SegmentDTO implements Serializable {
+
+    private static final long serialVersionUID = 4767453140231279111L;
 
     private long beginId;
 

im-logic-api/im-segment-id-api/src/main/java/net/sopod/soim/logic/segmentid/api/service/SegmentIdService.java

@@ -0,0 +1,18 @@
+package net.sopod.soim.logic.segmentid.api.service;
+
+import net.sopod.soim.logic.segmentid.api.model.dto.NextSegmentParam;
+import net.sopod.soim.logic.segmentid.api.model.dto.SegmentDTO;
+
+/**
+ * SegmentIdService
+ *
+ * @author tmy
+ * @date 2022-04-02 23:07
+ */
+public interface SegmentIdService {
+
+    SegmentDTO nextSegmentId(String bizTag);
+
+    SegmentDTO nextSegmentId(String bizTag, Long step);
+
+}

im-logic/im-segment-id/pom.xml

@@ -74,5 +74,17 @@
             <groupId>org.springframework</groupId>
             <artifactId>spring-orm</artifactId>
         </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.junit.vintage</groupId>
+                    <artifactId>junit-vintage-engine</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
     </dependencies>
 </project>

im-logic/im-segment-id/src/main/java/net/sopod/soim/logic/segmentid/SegmentIdApplication.java

@@ -1,6 +1,7 @@
 package net.sopod.soim.logic.segmentid;
 
 import org.apache.dubbo.config.spring.context.annotation.EnableDubbo;
+import org.mybatis.spring.annotation.MapperScan;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 
@@ -11,6 +12,7 @@ import org.springframework.boot.autoconfigure.SpringBootApplication;
  * @date 2022-04-02 14:55
  */
 @EnableDubbo(scanBasePackages = {"net.sopod.soim.logic.segmentid.service"})
+@MapperScan("net.sopod.soim.logic.segmentid.mapper")
 @SpringBootApplication
 public class SegmentIdApplication {
 

im-logic/im-segment-id/src/main/java/net/sopod/soim/logic/segmentid/config/SegmentConfig.java

@@ -23,6 +23,9 @@ public class SegmentConfig {
 
     private long initStep = 1000L;
 
-    private int dbSegmentVersionRetryTimes = 3;
+    //private int dbSegmentVersionRetryTimes = 10;
+
+    /** 从数据库查询区段号超时时间，并发时进行轮询 */
+    private long getSegmentPollDBTimeout = 2000L;
 
 }

im-logic/im-segment-id/src/main/java/net/sopod/soim/logic/segmentid/mapper/SegmentIdMapper.java

@@ -2,6 +2,7 @@ package net.sopod.soim.logic.segmentid.mapper;
 
 import com.baomidou.mybatisplus.core.mapper.BaseMapper;
 import net.sopod.soim.logic.segmentid.model.entity.SegmentId;
+import org.springframework.stereotype.Repository;
 
 /**
  * SegmentIdMapper
@@ -9,6 +10,7 @@ import net.sopod.soim.logic.segmentid.model.entity.SegmentId;
  * @author tmy
  * @date 2022-04-02 15:00
  */
+@Repository
 public interface SegmentIdMapper extends BaseMapper<SegmentId> {
 
 }

im-logic/im-segment-id/src/main/java/net/sopod/soim/logic/segmentid/service/SegmentIdServiceImpl.java

@@ -6,10 +6,10 @@ import com.google.common.base.Preconditions;
 import lombok.AllArgsConstructor;
 import net.sopod.soim.common.util.ImClock;
 import net.sopod.soim.common.util.ObjectUtil;
+import net.sopod.soim.logic.segmentid.api.model.dto.SegmentDTO;
+import net.sopod.soim.logic.segmentid.api.service.SegmentIdService;
 import net.sopod.soim.logic.segmentid.config.SegmentConfig;
 import net.sopod.soim.logic.segmentid.mapper.SegmentIdMapper;
-import net.sopod.soim.logic.segmentid.model.dto.NextSegmentParam;
-import net.sopod.soim.logic.segmentid.model.dto.SegmentDTO;
 import net.sopod.soim.logic.segmentid.model.entity.SegmentId;
 import net.sopod.soim.logic.segmentid.util.RedisLockUtil;
 import org.apache.dubbo.config.annotation.DubboService;
@@ -27,9 +27,9 @@ import java.util.concurrent.ThreadLocalRandom;
  */
 @DubboService
 @AllArgsConstructor
-public class SegmentIdService {
+public class SegmentIdServiceImpl implements SegmentIdService {
 
-    private static final Logger logger = LoggerFactory.getLogger(SegmentIdService.class);
+    private static final Logger logger = LoggerFactory.getLogger(SegmentIdServiceImpl.class);
 
     private final SegmentIdMapper segmentIdMapper;
 
@@ -37,8 +37,13 @@ public class SegmentIdService {
 
     private final SegmentConfig segmentConfig;
 
-    public SegmentDTO nextSegmentId(NextSegmentParam param) {
+    @Override
-        String bizTag = param.getBizTag();
+    public SegmentDTO nextSegmentId(String bizTag) {
+        return nextSegmentId(bizTag, null);
+    }
+
+    @Override
+    public SegmentDTO nextSegmentId(String bizTag, Long step) {
         SegmentId segmentId = segmentIdMapper.selectById(bizTag);
         if (segmentId == null) {
             String key = SegmentConfig.KEY_PREFIX_SEGMENT_ID_INSERT + bizTag;
@@ -53,7 +58,7 @@ public class SegmentIdService {
                             .setCreateTime(ImClock.date())
                             .setCurrentId(segmentConfig.getInitId())
                             .setVersion(0L)
-                            .setInitStep(segmentConfig.getInitStep());
+                            .setInitStep(ObjectUtil.defaultValue(step, segmentConfig.getInitStep()));
                     segmentIdMapper.insert(newSegmentId);
                     segmentId = newSegmentId;
                 } finally {
@@ -61,6 +66,7 @@ public class SegmentIdService {
                     RedisLockUtil.releaseLock(redisTemplate, key, val);
                 }
             } else {
+                logger.info("wait lock...");
                 // 等待锁释放
                 long timeout = 2000L;
                 boolean released = waitReleaseRedisLock(key, timeout);
@@ -73,15 +79,27 @@ public class SegmentIdService {
                 throw new RuntimeException(String.format("%s分段数据不存在,新增失败", bizTag));
             }
         }
-        int i = 0;
+        long begin = ImClock.millis();
-        do {
+        for (int i = 0; ; i++) {
-            SegmentDTO segment = getSegment(segmentId, param.getStep());
+            if (i > 0) {
+                segmentId = segmentIdMapper.selectById(bizTag);
+            }
+            SegmentDTO segment = getSegment(segmentId, step);
             if (segment != null) {
+                if (i > 0)
+                    logger.info("{} get segment try {} times", bizTag, i);
                 return segment;
             }
-            i++;
+            try {
-        } while (i < segmentConfig.getDbSegmentVersionRetryTimes());
+                Thread.sleep(ThreadLocalRandom.current().nextInt(50) + 50L);
-
+            } catch (InterruptedException e) {
+                logger.error("wait db interrupted!", e);
+            }
+            long current = ImClock.millis();
+            if (current - begin > segmentConfig.getGetSegmentPollDBTimeout()) {
+                break;
+            }
+        }
         throw new RuntimeException("id分段获取失败");
     }
 
@@ -106,11 +124,12 @@ public class SegmentIdService {
 
     /**
      * 获取id段
+     *
      * @param dbSegmentId 当前DB segmentId
-     * @param step 步长
+     * @param step        步长
      */
     private SegmentDTO getSegment(SegmentId dbSegmentId, Long step) {
-        step = ObjectUtil.defaultValue(step, dbSegmentId.getInitStep(), null);
+        step = ObjectUtil.defaultValue(step, dbSegmentId.getInitStep(), dbSegmentId.getInitStep());
         Preconditions.checkState(step > 0, "步长需大于0");
         // 开始id, 结束id
         Long currentId = dbSegmentId.getCurrentId();

im-logic/im-segment-id/src/main/resources/application.yml

@@ -7,7 +7,8 @@ spring:
     username: root
     password: sopod@2347#
   redis:
-    host: 124.222.131.236:3379
+    host: 124.222.131.236
+    port: 3379
     password: sopod@redis#
 
 dubbo:
@@ -20,3 +21,4 @@ dubbo:
 
 mybatis-plus:
   mapper-locations: classpath:mapper/*.xml
+

im-logic/im-segment-id/src/test/java/net/sopod/soim/logic/segmentid/test/SegmentIdApplicationTest.java

@@ -0,0 +1,48 @@
+package net.sopod.soim.logic.segmentid.test;
+
+import net.sopod.soim.common.util.ImClock;
+import net.sopod.soim.logic.segmentid.SegmentIdApplication;
+import net.sopod.soim.logic.segmentid.api.model.dto.SegmentDTO;
+import net.sopod.soim.logic.segmentid.api.service.SegmentIdService;
+import org.apache.dubbo.config.annotation.DubboReference;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+import java.util.concurrent.CountDownLatch;
+
+/**
+ * SegmentIdApplicationTest
+ *
+ * @author tmy
+ * @date 2022-04-03 00:54
+ */
+@RunWith(SpringJUnit4ClassRunner.class)
+@SpringBootTest(classes = SegmentIdApplication.class)
+public class SegmentIdApplicationTest {
+
+    @DubboReference
+    private SegmentIdService segmentIdService;
+
+    @Test
+    public void test1() throws InterruptedException {
+        CountDownLatch countDown = new CountDownLatch(10);
+        for (int i = 0; i < 10; i++) {
+            final int idx = i;
+            new Thread(() -> {
+                try {
+                    long begin = ImClock.millis();
+                    SegmentDTO segment = segmentIdService.nextSegmentId("im-user");
+                    System.out.println(idx + segment.toString() + (ImClock.millis() - begin));
+                }catch (Exception e){
+                    System.out.println(idx + "fail:" + e.getMessage());
+                } finally {
+                    countDown.countDown();
+                }
+            }).start();
+        }
+        countDown.await();
+    }
+
+}

pom.xml

@@ -129,6 +129,11 @@
                 <artifactId>spring-boot-starter-data-redis</artifactId>
                 <version>${spring-boot.version}</version>
             </dependency>
+            <dependency>
+                <groupId>io.lettuce</groupId>
+                <artifactId>lettuce-core</artifactId>
+                <version>6.1.6.RELEASE</version>
+            </dependency>
             <dependency>
                 <groupId>org.apache.dubbo</groupId>
                 <artifactId>dubbo-spring-boot-starter</artifactId>