From bfdd124ec68950944d210e8bbaa3be0da27a862e Mon Sep 17 00:00:00 2001 From: "zhipeng.zhang" <345849402@qq.com> Date: Fri, 6 Sep 2019 13:19:36 +0800 Subject: [PATCH] =?UTF-8?q?:ambulance:=E5=BF=85=E9=A1=BB=E4=BD=BF=E7=94=A8?= =?UTF-8?q?JDK1.8=E4=BB=A5=E4=B8=8A=E7=89=88=E6=9C=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 2 +- spring-boot-seckill.iml | 169 ++++++++ .../seckill/common/limit/NginxLimit.java | 138 ++++--- .../seckill/web/SeckillController.java | 118 +++--- .../web/SeckillDistributedController.java | 381 +++++++++--------- 5 files changed, 468 insertions(+), 340 deletions(-) create mode 100644 spring-boot-seckill.iml diff --git a/README.md b/README.md index 75ef3e7..e5f75a8 100644 --- a/README.md +++ b/README.md @@ -18,7 +18,7 @@ ## 开发环境 -JDK1.7、Maven、Mysql、Eclipse、SpringBoot1.5.10、zookeeper3.4.6、kafka_2.11、redis-2.8.4、curator-2.10.0 +JDK1.8、Maven、Mysql、IntelliJ IDEA、SpringBoot1.5.10、zookeeper3.4.6、kafka_2.11、redis-2.8.4、curator-2.10.0 ## 启动说明 diff --git a/spring-boot-seckill.iml b/spring-boot-seckill.iml new file mode 100644 index 0000000..708d20c --- /dev/null +++ b/spring-boot-seckill.iml @@ -0,0 +1,169 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/src/main/java/com/itstyle/seckill/common/limit/NginxLimit.java b/src/main/java/com/itstyle/seckill/common/limit/NginxLimit.java index 3962b7f..9ae1cdc 100644 --- a/src/main/java/com/itstyle/seckill/common/limit/NginxLimit.java +++ b/src/main/java/com/itstyle/seckill/common/limit/NginxLimit.java @@ -1,70 +1,68 @@ -package com.itstyle.seckill.common.limit; - -import java.io.BufferedReader; -import java.io.IOException; -import java.io.InputStreamReader; -import java.net.URL; -import java.net.URLConnection; -import java.util.concurrent.CountDownLatch; -/** - * Nginx 限流压测 - * src/mian/resources/nginx - * src/mian/resources/openresty - */ -public class NginxLimit { - //也可以使用AB压测 ab -n1000 -c 10 http://121.42.155.213 - private static int count = 0; - public static void main(String[] args) throws IOException, InterruptedException { - final CountDownLatch latch = new CountDownLatch(1); - for (int i = 0; i < 80; i++) { - Thread t = new Thread(new Runnable() { - public void run() { - try { - latch.await(); - String result = NginxLimit.sendGet("http://121.42.155.213"); - System.out.println(result); - } catch (InterruptedException e) { - e.printStackTrace(); - } - } - }); - t.start(); - } - latch.countDown(); - Thread.sleep(5000); - System.out.println(count); - // System.in.read(); - } - //发送GET请求 - public static String sendGet(String url) { - String result = ""; - BufferedReader in = null; - try { - String urlName = url; - URL realUrl = new URL(urlName); - URLConnection conn = realUrl.openConnection();// 打开和URL之间的连接 - conn.setRequestProperty("accept", "*/*");// 设置通用的请求属性 - conn.setRequestProperty("connection", "Keep-Alive"); - conn.setRequestProperty("user-agent", "Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1)"); - conn.setConnectTimeout(4000); - conn.connect();// 建立实际的连接 - in = new BufferedReader(new InputStreamReader(conn.getInputStream(), "UTF-8"));// 定义BufferedReader输入流来读取URL的响应 - String line; - while ((line = in.readLine()) != null) { - result += line; - } - } catch (Exception e) { - count ++; - System.out.println("发送GET请求出现异常！" + e); - } finally {// 使用finally块来关闭输入流 - try { - if (in != null) { - in.close(); - } - } catch (IOException ex) { - System.out.println("关闭流异常"); - } - } - return result; - } -} +package com.itstyle.seckill.common.limit; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStreamReader; +import java.net.URL; +import java.net.URLConnection; +import java.util.concurrent.CountDownLatch; +/** + * Nginx 限流压测 + * src/mian/resources/nginx + * src/mian/resources/openresty + */ +public class NginxLimit { + //也可以使用AB压测 ab -n1000 -c 10 http://121.42.155.213 + private static int count = 0; + public static void main(String[] args) throws IOException, InterruptedException { + final CountDownLatch latch = new CountDownLatch(1); + for (int i = 0; i < 80; i++) { + Thread t = new Thread(() -> { + try { + latch.await(); + String result = NginxLimit.sendGet("http://121.42.155.213"); + System.out.println(result); + } catch (InterruptedException e) { + e.printStackTrace(); + } + }); + t.start(); + } + latch.countDown(); + Thread.sleep(5000); + System.out.println(count); + // System.in.read(); + } + //发送GET请求 + public static String sendGet(String url) { + String result = ""; + BufferedReader in = null; + try { + String urlName = url; + URL realUrl = new URL(urlName); + URLConnection conn = realUrl.openConnection();// 打开和URL之间的连接 + conn.setRequestProperty("accept", "*/*");// 设置通用的请求属性 + conn.setRequestProperty("connection", "Keep-Alive"); + conn.setRequestProperty("user-agent", "Mozilla/4.0 (compatible; MSIE 6.0; Windows NT 5.1; SV1)"); + conn.setConnectTimeout(4000); + conn.connect();// 建立实际的连接 + in = new BufferedReader(new InputStreamReader(conn.getInputStream(), "UTF-8"));// 定义BufferedReader输入流来读取URL的响应 + String line; + while ((line = in.readLine()) != null) { + result += line; + } + } catch (Exception e) { + count ++; + System.out.println("发送GET请求出现异常！" + e); + } finally {// 使用finally块来关闭输入流 + try { + if (in != null) { + in.close(); + } + } catch (IOException ex) { + System.out.println("关闭流异常"); + } + } + return result; + } +} diff --git a/src/main/java/com/itstyle/seckill/web/SeckillController.java b/src/main/java/com/itstyle/seckill/web/SeckillController.java index 15752e6..24dbe15 100644 --- a/src/main/java/com/itstyle/seckill/web/SeckillController.java +++ b/src/main/java/com/itstyle/seckill/web/SeckillController.java @@ -33,7 +33,7 @@ public class SeckillController { private static int corePoolSize = Runtime.getRuntime().availableProcessors(); //创建线程池 调整队列数 拒绝服务 private static ThreadPoolExecutor executor = new ThreadPoolExecutor(corePoolSize, corePoolSize+1, 10l, TimeUnit.SECONDS, - new LinkedBlockingQueue(1000)); + new LinkedBlockingQueue<>(1000)); @Autowired private ISeckillService seckillService; @@ -55,17 +55,14 @@ public class SeckillController { RequestContextHolder.setRequestAttributes(sra, true); for(int i=0;i { + Result result = seckillService.startSeckil(killId, userId); + if(result!=null){ + LOGGER.info("用户:{}{}",userId,result.get("msg")); + }else{ + LOGGER.info("用户:{}{}",userId,"哎呦喂，人也太多了，请稍后！"); } + latch.countDown(); }; executor.execute(task); } @@ -88,13 +85,10 @@ public class SeckillController { LOGGER.info("开始秒杀二(正常)"); for(int i=0;i<1000;i++){ final long userId = i; - Runnable task = new Runnable() { - @Override - public void run() { - Result result = seckillService.startSeckilLock(killId, userId); - LOGGER.info("用户:{}{}",userId,result.get("msg")); - latch.countDown(); - } + Runnable task = () -> { + Result result = seckillService.startSeckilLock(killId, userId); + LOGGER.info("用户:{}{}",userId,result.get("msg")); + latch.countDown(); }; executor.execute(task); } @@ -117,13 +111,10 @@ public class SeckillController { LOGGER.info("开始秒杀三(正常)"); for(int i=0;i<1000;i++){ final long userId = i; - Runnable task = new Runnable() { - @Override - public void run() { - Result result = seckillService.startSeckilAopLock(killId, userId); - LOGGER.info("用户:{}{}",userId,result.get("msg")); - latch.countDown(); - } + Runnable task = () -> { + Result result = seckillService.startSeckilAopLock(killId, userId); + LOGGER.info("用户:{}{}",userId,result.get("msg")); + latch.countDown(); }; executor.execute(task); } @@ -146,13 +137,10 @@ public class SeckillController { LOGGER.info("开始秒杀四(正常)"); for(int i=0;i<1000;i++){ final long userId = i; - Runnable task = new Runnable() { - @Override - public void run() { - Result result = seckillService.startSeckilDBPCC_ONE(killId, userId); - LOGGER.info("用户:{}{}",userId,result.get("msg")); - latch.countDown(); - } + Runnable task = () -> { + Result result = seckillService.startSeckilDBPCC_ONE(killId, userId); + LOGGER.info("用户:{}{}",userId,result.get("msg")); + latch.countDown(); }; executor.execute(task); } @@ -175,13 +163,10 @@ public class SeckillController { LOGGER.info("开始秒杀五(正常、数据库锁最优实现)"); for(int i=0;i<1000;i++){ final long userId = i; - Runnable task = new Runnable() { - @Override - public void run() { - Result result = seckillService.startSeckilDBPCC_TWO(killId, userId); - LOGGER.info("用户:{}{}",userId,result.get("msg")); - latch.countDown(); - } + Runnable task = () -> { + Result result = seckillService.startSeckilDBPCC_TWO(killId, userId); + LOGGER.info("用户:{}{}",userId,result.get("msg")); + latch.countDown(); }; executor.execute(task); } @@ -204,15 +189,12 @@ public class SeckillController { LOGGER.info("开始秒杀六(正常、数据库锁最优实现)"); for(int i=0;i<1000;i++){ final long userId = i; - Runnable task = new Runnable() { - @Override - public void run() { - //这里使用的乐观锁、可以自定义抢购数量、如果配置的抢购人数比较少、比如120:100(人数:商品) 会出现少买的情况 - //用户同时进入会出现更新失败的情况 - Result result = seckillService.startSeckilDBOCC(killId, userId,1); - LOGGER.info("用户:{}{}",userId,result.get("msg")); - latch.countDown(); - } + Runnable task = () -> { + //这里使用的乐观锁、可以自定义抢购数量、如果配置的抢购人数比较少、比如120:100(人数:商品) 会出现少买的情况 + //用户同时进入会出现更新失败的情况 + Result result = seckillService.startSeckilDBOCC(killId, userId,1); + LOGGER.info("用户:{}{}",userId,result.get("msg")); + latch.countDown(); }; executor.execute(task); } @@ -233,23 +215,20 @@ public class SeckillController { LOGGER.info("开始秒杀柒(正常)"); for(int i=0;i<1000;i++){ final long userId = i; - Runnable task = new Runnable() { - @Override - public void run() { - SuccessKilled kill = new SuccessKilled(); - kill.setSeckillId(killId); - kill.setUserId(userId); - try { - Boolean flag = SeckillQueue.getMailQueue().produce(kill); - if(flag){ - LOGGER.info("用户:{}{}",kill.getUserId(),"秒杀成功"); - }else{ - LOGGER.info("用户:{}{}",userId,"秒杀失败"); - } - } catch (InterruptedException e) { - e.printStackTrace(); + Runnable task = () -> { + SuccessKilled kill = new SuccessKilled(); + kill.setSeckillId(killId); + kill.setUserId(userId); + try { + Boolean flag = SeckillQueue.getMailQueue().produce(kill); + if(flag){ + LOGGER.info("用户:{}{}",kill.getUserId(),"秒杀成功"); + }else{ LOGGER.info("用户:{}{}",userId,"秒杀失败"); } + } catch (InterruptedException e) { + e.printStackTrace(); + LOGGER.info("用户:{}{}",userId,"秒杀失败"); } }; executor.execute(task); @@ -271,14 +250,11 @@ public class SeckillController { LOGGER.info("开始秒杀八(正常)"); for(int i=0;i<1000;i++){ final long userId = i; - Runnable task = new Runnable() { - @Override - public void run() { - SeckillEvent kill = new SeckillEvent(); - kill.setSeckillId(killId); - kill.setUserId(userId); - DisruptorUtil.producer(kill); - } + Runnable task = () -> { + SeckillEvent kill = new SeckillEvent(); + kill.setSeckillId(killId); + kill.setUserId(userId); + DisruptorUtil.producer(kill); }; executor.execute(task); } diff --git a/src/main/java/com/itstyle/seckill/web/SeckillDistributedController.java b/src/main/java/com/itstyle/seckill/web/SeckillDistributedController.java index 506e413..0e56a5d 100644 --- a/src/main/java/com/itstyle/seckill/web/SeckillDistributedController.java +++ b/src/main/java/com/itstyle/seckill/web/SeckillDistributedController.java @@ -1,198 +1,183 @@ -package com.itstyle.seckill.web; -import io.swagger.annotations.Api; -import io.swagger.annotations.ApiOperation; - -import java.util.concurrent.LinkedBlockingQueue; -import java.util.concurrent.ThreadPoolExecutor; -import java.util.concurrent.TimeUnit; - -import javax.jms.Destination; - -import org.apache.activemq.command.ActiveMQQueue; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.web.bind.annotation.PostMapping; -import org.springframework.web.bind.annotation.RequestMapping; -import org.springframework.web.bind.annotation.RestController; - -import com.itstyle.seckill.common.entity.Result; -import com.itstyle.seckill.common.redis.RedisUtil; -import com.itstyle.seckill.queue.activemq.ActiveMQSender; -import com.itstyle.seckill.queue.kafka.KafkaSender; -import com.itstyle.seckill.queue.redis.RedisSender; -import com.itstyle.seckill.service.ISeckillDistributedService; -import com.itstyle.seckill.service.ISeckillService; -@Api(tags ="分布式秒杀") -@RestController -@RequestMapping("/seckillDistributed") -public class SeckillDistributedController { - private final static Logger LOGGER = LoggerFactory.getLogger(SeckillDistributedController.class); - - private static int corePoolSize = Runtime.getRuntime().availableProcessors(); - //调整队列数 拒绝服务 - private static ThreadPoolExecutor executor = new ThreadPoolExecutor(corePoolSize, corePoolSize+1, 10l, TimeUnit.SECONDS, - new LinkedBlockingQueue(10000)); - - @Autowired - private ISeckillService seckillService; - @Autowired - private ISeckillDistributedService seckillDistributedService; - @Autowired - private RedisSender redisSender; - @Autowired - private KafkaSender kafkaSender; - @Autowired - private ActiveMQSender activeMQSender; - - @Autowired - private RedisUtil redisUtil; - - @ApiOperation(value="秒杀一(Rediss分布式锁)",nickname="科帮网") - @PostMapping("/startRedisLock") - public Result startRedisLock(long seckillId){ - seckillService.deleteSeckill(seckillId); - final long killId = seckillId; - LOGGER.info("开始秒杀一"); - for(int i=0;i<1000;i++){ - final long userId = i; - Runnable task = new Runnable() { - @Override - public void run() { - Result result = seckillDistributedService.startSeckilRedisLock(killId, userId); - LOGGER.info("用户:{}{}",userId,result.get("msg")); - } - }; - executor.execute(task); - } - try { - Thread.sleep(15000); - Long seckillCount = seckillService.getSeckillCount(seckillId); - LOGGER.info("一共秒杀出{}件商品",seckillCount); - } catch (InterruptedException e) { - e.printStackTrace(); - } - return Result.ok(); - } - @ApiOperation(value="秒杀二(zookeeper分布式锁)",nickname="科帮网") - @PostMapping("/startZkLock") - public Result startZkLock(long seckillId){ - seckillService.deleteSeckill(seckillId); - final long killId = seckillId; - LOGGER.info("开始秒杀二"); - for(int i=0;i<10000;i++){ - final long userId = i; - Runnable task = new Runnable() { - @Override - public void run() { - Result result = seckillDistributedService.startSeckilZksLock(killId, userId); - LOGGER.info("用户:{}{}",userId,result.get("msg")); - } - }; - executor.execute(task); - } - try { - Thread.sleep(10000); - Long seckillCount = seckillService.getSeckillCount(seckillId); - LOGGER.info("一共秒杀出{}件商品",seckillCount); - } catch (InterruptedException e) { - e.printStackTrace(); - } - return Result.ok(); - } - @ApiOperation(value="秒杀三(Redis分布式队列-订阅监听)",nickname="科帮网") - @PostMapping("/startRedisQueue") - public Result startRedisQueue(long seckillId){ - redisUtil.cacheValue(seckillId+"", null);//秒杀结束 - seckillService.deleteSeckill(seckillId); - final long killId = seckillId; - LOGGER.info("开始秒杀三"); - for(int i=0;i<1000;i++){ - final long userId = i; - Runnable task = new Runnable() { - @Override - public void run() { - if(redisUtil.getValue(killId+"")==null){ - //思考如何返回给用户信息ws - redisSender.sendChannelMess("seckill",killId+";"+userId); - }else{ - //秒杀结束 - } - } - }; - executor.execute(task); - } - try { - Thread.sleep(10000); - redisUtil.cacheValue(killId+"", null); - Long seckillCount = seckillService.getSeckillCount(seckillId); - LOGGER.info("一共秒杀出{}件商品",seckillCount); - } catch (InterruptedException e) { - e.printStackTrace(); - } - return Result.ok(); - } - @ApiOperation(value="秒杀四(Kafka分布式队列)",nickname="科帮网") - @PostMapping("/startKafkaQueue") - public Result startKafkaQueue(long seckillId){ - seckillService.deleteSeckill(seckillId); - final long killId = seckillId; - LOGGER.info("开始秒杀四"); - for(int i=0;i<1000;i++){ - final long userId = i; - Runnable task = new Runnable() { - @Override - public void run() { - if(redisUtil.getValue(killId+"")==null){ - //思考如何返回给用户信息ws - kafkaSender.sendChannelMess("seckill",killId+";"+userId); - }else{ - //秒杀结束 - } - } - }; - executor.execute(task); - } - try { - Thread.sleep(10000); - redisUtil.cacheValue(killId+"", null); - Long seckillCount = seckillService.getSeckillCount(seckillId); - LOGGER.info("一共秒杀出{}件商品",seckillCount); - } catch (InterruptedException e) { - e.printStackTrace(); - } - return Result.ok(); - } - @ApiOperation(value="秒杀五(ActiveMQ分布式队列)",nickname="科帮网") - @PostMapping("/startActiveMQQueue") - public Result startActiveMQQueue(long seckillId){ - seckillService.deleteSeckill(seckillId); - final long killId = seckillId; - LOGGER.info("开始秒杀五"); - for(int i=0;i<1000;i++){ - final long userId = i; - Runnable task = new Runnable() { - @Override - public void run() { - if(redisUtil.getValue(killId+"")==null){ - Destination destination = new ActiveMQQueue("seckill.queue"); - //思考如何返回给用户信息ws - activeMQSender.sendChannelMess(destination,killId+";"+userId); - }else{ - //秒杀结束 - } - } - }; - executor.execute(task); - } - try { - Thread.sleep(10000); - redisUtil.cacheValue(killId+"", null); - Long seckillCount = seckillService.getSeckillCount(seckillId); - LOGGER.info("一共秒杀出{}件商品",seckillCount); - } catch (InterruptedException e) { - e.printStackTrace(); - } - return Result.ok(); - } -} +package com.itstyle.seckill.web; +import io.swagger.annotations.Api; +import io.swagger.annotations.ApiOperation; + +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +import javax.jms.Destination; + +import org.apache.activemq.command.ActiveMQQueue; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; + +import com.itstyle.seckill.common.entity.Result; +import com.itstyle.seckill.common.redis.RedisUtil; +import com.itstyle.seckill.queue.activemq.ActiveMQSender; +import com.itstyle.seckill.queue.kafka.KafkaSender; +import com.itstyle.seckill.queue.redis.RedisSender; +import com.itstyle.seckill.service.ISeckillDistributedService; +import com.itstyle.seckill.service.ISeckillService; +@Api(tags ="分布式秒杀") +@RestController +@RequestMapping("/seckillDistributed") +public class SeckillDistributedController { + private final static Logger LOGGER = LoggerFactory.getLogger(SeckillDistributedController.class); + + private static int corePoolSize = Runtime.getRuntime().availableProcessors(); + //调整队列数 拒绝服务 + private static ThreadPoolExecutor executor = new ThreadPoolExecutor(corePoolSize, corePoolSize+1, 10l, TimeUnit.SECONDS, + new LinkedBlockingQueue<>(10000)); + + @Autowired + private ISeckillService seckillService; + @Autowired + private ISeckillDistributedService seckillDistributedService; + @Autowired + private RedisSender redisSender; + @Autowired + private KafkaSender kafkaSender; + @Autowired + private ActiveMQSender activeMQSender; + + @Autowired + private RedisUtil redisUtil; + + @ApiOperation(value="秒杀一(Rediss分布式锁)",nickname="科帮网") + @PostMapping("/startRedisLock") + public Result startRedisLock(long seckillId){ + seckillService.deleteSeckill(seckillId); + final long killId = seckillId; + LOGGER.info("开始秒杀一"); + for(int i=0;i<1000;i++){ + final long userId = i; + Runnable task = () -> { + Result result = seckillDistributedService.startSeckilRedisLock(killId, userId); + LOGGER.info("用户:{}{}",userId,result.get("msg")); + }; + executor.execute(task); + } + try { + Thread.sleep(15000); + Long seckillCount = seckillService.getSeckillCount(seckillId); + LOGGER.info("一共秒杀出{}件商品",seckillCount); + } catch (InterruptedException e) { + e.printStackTrace(); + } + return Result.ok(); + } + @ApiOperation(value="秒杀二(zookeeper分布式锁)",nickname="科帮网") + @PostMapping("/startZkLock") + public Result startZkLock(long seckillId){ + seckillService.deleteSeckill(seckillId); + final long killId = seckillId; + LOGGER.info("开始秒杀二"); + for(int i=0;i<10000;i++){ + final long userId = i; + Runnable task = () -> { + Result result = seckillDistributedService.startSeckilZksLock(killId, userId); + LOGGER.info("用户:{}{}",userId,result.get("msg")); + }; + executor.execute(task); + } + try { + Thread.sleep(10000); + Long seckillCount = seckillService.getSeckillCount(seckillId); + LOGGER.info("一共秒杀出{}件商品",seckillCount); + } catch (InterruptedException e) { + e.printStackTrace(); + } + return Result.ok(); + } + @ApiOperation(value="秒杀三(Redis分布式队列-订阅监听)",nickname="科帮网") + @PostMapping("/startRedisQueue") + public Result startRedisQueue(long seckillId){ + redisUtil.cacheValue(seckillId+"", null);//秒杀结束 + seckillService.deleteSeckill(seckillId); + final long killId = seckillId; + LOGGER.info("开始秒杀三"); + for(int i=0;i<1000;i++){ + final long userId = i; + Runnable task = () -> { + if(redisUtil.getValue(killId+"")==null){ + //思考如何返回给用户信息ws + redisSender.sendChannelMess("seckill",killId+";"+userId); + }else{ + //秒杀结束 + } + }; + executor.execute(task); + } + try { + Thread.sleep(10000); + redisUtil.cacheValue(killId+"", null); + Long seckillCount = seckillService.getSeckillCount(seckillId); + LOGGER.info("一共秒杀出{}件商品",seckillCount); + } catch (InterruptedException e) { + e.printStackTrace(); + } + return Result.ok(); + } + @ApiOperation(value="秒杀四(Kafka分布式队列)",nickname="科帮网") + @PostMapping("/startKafkaQueue") + public Result startKafkaQueue(long seckillId){ + seckillService.deleteSeckill(seckillId); + final long killId = seckillId; + LOGGER.info("开始秒杀四"); + for(int i=0;i<1000;i++){ + final long userId = i; + Runnable task = () -> { + if(redisUtil.getValue(killId+"")==null){ + //思考如何返回给用户信息ws + kafkaSender.sendChannelMess("seckill",killId+";"+userId); + }else{ + //秒杀结束 + } + }; + executor.execute(task); + } + try { + Thread.sleep(10000); + redisUtil.cacheValue(killId+"", null); + Long seckillCount = seckillService.getSeckillCount(seckillId); + LOGGER.info("一共秒杀出{}件商品",seckillCount); + } catch (InterruptedException e) { + e.printStackTrace(); + } + return Result.ok(); + } + @ApiOperation(value="秒杀五(ActiveMQ分布式队列)",nickname="科帮网") + @PostMapping("/startActiveMQQueue") + public Result startActiveMQQueue(long seckillId){ + seckillService.deleteSeckill(seckillId); + final long killId = seckillId; + LOGGER.info("开始秒杀五"); + for(int i=0;i<1000;i++){ + final long userId = i; + Runnable task = () -> { + if(redisUtil.getValue(killId+"")==null){ + Destination destination = new ActiveMQQueue("seckill.queue"); + //思考如何返回给用户信息ws + activeMQSender.sendChannelMess(destination,killId+";"+userId); + }else{ + //秒杀结束 + } + }; + executor.execute(task); + } + try { + Thread.sleep(10000); + redisUtil.cacheValue(killId+"", null); + Long seckillCount = seckillService.getSeckillCount(seckillId); + LOGGER.info("一共秒杀出{}件商品",seckillCount); + } catch (InterruptedException e) { + e.printStackTrace(); + } + return Result.ok(); + } +}