From 2f49bed8611517ea63e42848db069075e08649da Mon Sep 17 00:00:00 2001 From: Toyas Date: Thu, 10 Apr 2025 04:14:49 +0800 Subject: [PATCH] fix(content): Improve video statistics messaging fallback --- .idea/.cache/.Apifox_Helper/.toolWindow.db | Bin 294912 -> 303104 bytes .../video/content/config/RocketConfig.java | 17 ++++- .../mq/producer/VideoMessageProducer.java | 62 +++++++++++++++--- .../service/impl/VideoServiceImpl.java | 18 +++-- 4 files changed, 77 insertions(+), 20 deletions(-) diff --git a/.idea/.cache/.Apifox_Helper/.toolWindow.db b/.idea/.cache/.Apifox_Helper/.toolWindow.db index fd168ed1de459bb1ca3f833f759f78d6dc8092c8..aa3ef4cc23432744bc85322c18441c64806bc0f5 100644 GIT binary patch delta 4843 zcmai%d301o_Q(5Hz3T34opgFh(pkDA-3bDL-WS#cWG5s9!jiD0SBMZ6Wsx;BI7E`D z=_yDl0>NbjWQ~F);4rx2h%o3d`ulNIkkJ`YVN{TDK*y&^9kxA1neUSN)MB=rVZdo2JYEHI_ zTZHzAn}psI&k$N1R|&PoL8y6^kfNnREDMEXM}%l+2+5ivL|-GMV7L&&5Fy5ZkQ|qg z+%h3~J%yNh2+1!LQkW}5mmEN&OcnZqD%`9v-%=gWypy{u>tW5b{I%Kn=GpmLlS|!{ zIovoIQ4YSV)e+vf!XDmr_vYX{59o`j4q8k&u*XvGjpoaKg<7J3dyZ$ zR&C7uR##LM(dOo@(C*DSp0PaV$@F8IE1B`s`n;~ZH&f$Ar29xeBJ+83Vcz8om1%13 z)}sAc*RszQ=#As@YEycrMa|v1Il32Bw{*jDUrU*gHLJ*ET$I*k7+v(|EJNxw!+g`p zT($aw{=7Dn(<|#t;SYtu{Q4Y{>d~yppYo4_)4E3s&#I3YK2e2DtM!Kq_vHR9Yq!25 z^(l-};pVi;knU`kzCbvxW7xU1xA(`136SXaA-Dy7?FPcA0&D6f1 zeLy=%lbP?;JgfVuJE2n2TO)1o0xce8BfD+560v!^!GaXKXp5|(ttjD6wc6wRxZ8UR?4YpFx7#n#? zjzVWLx`oCCf0FJL`UE{KqD+ofZNnC&qEq%Sb|qVGrbEWs$R?Q-3{MhD+s4}9x3pXE zgv=*gM}l|IpU2vIKb|DPn8zRU#q1%c$LWhk!!f(V7xG6up@`S*k42q|$K>Z)tCOQW zD*NV1$-dj^f^jy|CXWVAKucRG^6jr21Sd_)cwd6J}smQ~uw2AM1xm?XcU%PVd0 zFzph2z0CJ%lHIIJ`t)k0aN{!WC`jV9R$K`z>k_g+Go3cx3Tx<^@m8{0#%0$8UPa## zcqP3e@Cq50f8H2Jh%~C$Sm14 zFTxg>NsA_0$qX4JRtYej&Y5V1ditoqbuvENBzl@gKN5H<{YmH)T0F@Llj#VdwREn~ zNwi~oMc$kdKodOS~y{1^9 zf>uqjl5!dMa|=9#J}K~EdR*W^GXBH&A|(T5x9PITzyMk{)e8M-jnDvHCe%;23iZ*~ zrdo+tZg_CFXy~D7)2!g8_Gwn)lJV2;VM~cK=|%#cfH&d_L_?mKH)N06eUYF&=nZ&d zE~nET@HprS@vNP8iD&!C&uUMJF8k8U0+&%uot5;Fam9HtyWX<9CM;Q?7hNr4_uh^z z5}+qNS!ab(`i;;Ms;##YTQb4?zuRj)v`UBCJ#}{-2 zBR!IhVnKT-9Q21G?nopU4Y^znmopTIy6kpW)DiIo*4NeZx_WrwJ}#Yr%VsouIs3;2 zFf^oQcPNUE6zMZ3PXp!Tlc9hS3pt0gqejE39|^x7N!-JB}^}@K$t<8QCNA20fK9iEsNmR4QR2|N06XAo^q%O4zF~jm`a0 zH#Ebp8V!596HfqcGtJ=5i}6+h(|N%P{JH2Pvz}iMBCQhYxN;6UOkf&+wF%V#Oy!vi z(2K;F>`Voc!ZcIATXQwVf)^t*NQuN%ye>Dw^L~p=0AcQ?Xc9n(Z`_3T5D0S57Icc> zHh%+e3ge>$X7Qel_z;1a{FC{(i@aQpBuipRdEmP{q#w0AIp*_QnnP5i0qs{{%I_IR4g;&;T%& zm)wN&V*C+f4j)JF5E#u{l;|{pQM@)CO(QUpAJ8TzK7zlSjlyszW7`#`JpC?RO-73< z0o%J1ts-L!hj5b{eNSL8&*+b?i9SaS;=W7xT>=9+`V!Xy4B%5Q#P|^)nn1@?!-i~guwtmcpd+hfS+%=iRTgAI@`y$@4$@&yj=Y}78&&L z89VV#0&Y(B;g3nD1t~SJD@=a9R+Fi^plnyXo=EJ+Z6jU$^B3_S2{`%wS8z21+RGfq z$`);9Jjc+Ow_#Xi%e|$^9j#E@?Q>^|b9gfQMUB<$`JPro>rp* zcJ*DXVP}paGrN_FwU)b1#BmA_X5YPvoNQ|vu4AizkLuXjPw+8jAxO)qwFAVU{IQ20Nig>eO+K z4=%1=5}MZ-X5aNiMs{KZQYshT29^?B!Wu4tnl|@0@qI7`5maco%S0nAxeY9y zs5|8Igo9px!0C31%H($XVv&fr!vzA~kR#0Al%#oPzdJ0hbbgn|6>&JCF`v&Xu6wRX z%pVJc#I=tl{L&&OvYb%fdCiJ3$o%UYmiNMaSVJ08)3t+5OuS(=Yxoy3u<{Ftu=X-k z%XYjiUg+XZWMvkehCdDZm9fk@;5~9B>yfYidFPS1K7f&xFDJLRCR2@|Cja5 zeqtG_FM*j|Tp;;bYKpYm9B}*m0k1a_2>7FJ`QwX59G;LvWIiU06@Psv?W#LpAses? zm&tigm-F85&XN|VN=D0FrW&}N{+Ki5^+W?gf6ZeeeIf44*ks z9y{@{I3+C0@nHU8y42T|rBpnwFb&ZCZ}!UckaCXV@kC;!_;iPJyB_@yf%|xh3H?*l zF5@u1%7Sv0FqCgDK{Eg<__b0LR6;qg?TcPdp2Rz)a>A09eTvUq{CYtl+MjqUzxGs^$EP6KH}G z9^$@}NK(QwKJYX;Ml8t>Jsr*|O#Age=r(Dp)D6m?$TjJvq{VJg2hVjU4grp97P3cP zLLIPxyZfQw!!x0O`-5?fIDn&-ah0JyPn-oUtBY?Rz<2x4$rha!hd@Iq zHnNUMID>6&L7l8%mXylsBeRzav1lw5_WS*jfHN2qcVM^69d!Cbo>0)~ih8^r z29w13khT$pEq5Kx+*4H3_ckJj#pj85+)i=Rc6z<`fVftB+-{FkRB*Rn)OB%iS8NkA zo7#$|vIZ~qvdmJP&kAesau#kw+gbiBDJ?n2KI}xgWMpSwL}P9@IXey4>*P{JuuWlF zr+-^>DMgEI-3c9*+UAJc@hCn1ilDfHyG&vY#r>z5-~0@T`_CHw>EFfnht+(`6}0l# z4Rb+ZI%znbwIyw~QmeR-NSu;d`^GJN)nRZDF!TLKp@vAU{o+M@VjMdO6!Hxla1R0n Q{Dn5GCXmlx>cAfIALlA+Qvd(} delta 3439 zcmXY!d3Y2>8pivp?wXloGBe3sndw{;(i4slAR*jWIFkS&B!rNIBO!)cB#7LQ45wy5 z(-zRm6+i_PkR1dIMA?9-C=08IhKB`25Eoog5O-0cvfuQ>AK&}D&+q-Zy1J^nCaG_> z*Vo${<7f{-5ERun0D6QfUr#c*lU_t1SXq>>qpE)RPOEQyA}dHW)FDg=$f$7v8>$>I zqIv^>>RE~sJ0Hb82gOx`5T68@V|6;}~EFY=uA zZrsoD=Nx||d>&Eg&KqfLl|OMQ3H!WbV?55s)~v{GmZpSQb0~hV#Tfrf^qa9XuDfTe z>FK0F<`!GP@=?tHqEe!si7iO_I&zOa+mRTv-@4ppcg_r{dkm~P$4__r#eAZv7r{q_ zH`qZWuj{1z>o9qZwdVL?7fa6d)15khsU^&Ju&0qU>7@B8_zc2!-G6Z{oNi;^V7gT& zeV>;w-J<*YpoHmWR+#6fuj=G|Elgfv+mL8F`Cwp}Y+`rv{IHSPhWlXy>xRl$1?tNz zFx*d7z4QLaaAzYshvX%lq~(Xni_AO159`^05q`Q(=h>xU9%Lcp4Xg$A1@%=F9$treyq#stYZ;`Lic~f+l2UvQ(pVlj4K4?wxl8@Hu zwtqF+<+@Fqj&_-DTW6zPs@reoNj_M@Y@-yqSO*ywL?2zG+Y_fnA1q|cFrtShxln7-@l1?7VAA) zfthU5Xob$uar7;``*huA-^Pw<>^x>ub+9N5rZ7*T0yV5(p+cVuV|KmC2i0sX(#dQ; zY8Cqevr7Gv^(n+hC+W66jrd?9%Pvyr1Rea@9l?0kP^3TwJBYYk$0s8Ym$Bawm$Jk$ z3LU58+V8;!C2ZOl1;(=Vh>LYx_XGIo7~S5xiycMm59}yp$;ApCt>b=`I9>spS**Y) z)`&P?$4e&T%tq?AOAU66V2?2yu7elmAjo5RV-?6{v&SklN5^yK;w8g$+kYW;3}sib zV+bQ93LUKD$CGjO2I)311s7`|tHmr^2TiN->H)e{=VR8NUB|4S4x(0J)>pUCfFt)| z1>+RjTL+VNW7bQz%^NW5$xdL_LkD%c5OimzQU$uP^iqXp>A31BUeZ;!|9lraGTBz_ z$Y3Y2ql=D{uOaTtx|AujlMXsu#7ok3t6ssZBWo&CAdQ_y?ZED!wrAFIg|^dMw*P_M zZFM_|;<;04{iovpD`LGhPlCTu4WUMw=;ok`*q^<^jYbT@_q<|Di z)izv!wcyv{zl4uK(QbSR7a&DDauKdU8*Rf^a0imL-*X8lg+zYf zD%b!L_y^ZuFh~uZ0`dH2y_7*Aj=R=KMfBYra5sqz%+h}TJGf*s)CzKn=Tz)5ceAU- z`N)~(*lf9N+zhoLbqt8>Qj7TIyU@-6gFP}@!Q5eeGEhJ?WhVT{3hzu~8Z(B~PK@64*;xAW< zmnaP6GpC3T>8564ur3nhNN23I$av52vam>1Ys7W#Y+f@<^cY|OUp`ly0MMUbc~-14 zKtJApv3QD#!8v{T&H(v^LLa_iEqM?gN!lZ26JN9e?hA6Y^Q0}xa?B*c{g8SU7sXb} zyIM$$0mkv64swJ-32*g~B7m`cluQoch)KnK%MI8`VT{&n5?#FLHms(e@SgxCp+=B1 zoogH!c6ZciqnW-X)TnB{*f6Mu7yTp-q3{&f9*DSD)%>?##iJA^^Ux!)fQly-1B)9qMoxnb%?ibHC5 zA_gbU;TPT(uTyx2zjIjJPhmE%d{5jF-q5GDmT$#In59j-A)besT92FJ6_}wpzY}l6 z^zeo{G(1^v!8HEATO#i2RQ~vTu^M0suf8q*NS)!^8-yEzoa`*JKeY5RW(hY`^}M*Q zU_SrpoLFdp=Xmx-Q8K_hKJ>D993QFlS^nuxu^eD7S6&m(Q>f(+_KH`Yj0Fz`xr?*b z;j(DPf5F3$dK0(3YAL@pTe?YM3IBASbezIszIw5=1P3cyq}3WE7x$bLhXE|)lTL{z zC@kQKXT)>V9-bW~x#1CqS~r?zk=*ctNb7t7;0CLdVgP}6a7t|rP{+r}(gEDXY0G*0 zYUu|G%eZU0bneNlXo=uGrj%a}1K3|EH&v13-D# z-guc5GzO$2-%AY?M2<5Y4?y^Nqm*TM@<#L(~ijfmP~)(o$JD)d!VMOsl9u<_;* z5^l7WyYeLi1q+WalKzq)ipPzU@GCd-v{GpzNWq1HNWRi1wNi-Sb*WOS0Ze>%N9l9A zDO-}PRGcBmUguKlxtt#avXw#^ zA891tJ$bF5Ku98qo`hcUQ{r~Sd0e-n51MxyRzN{W&A?-(b3o=ZgHk31FW=rMrBLwj zQyV1{g;;)ev($}B4IKk+zIUM1nSx8ZG)!6xPOUOW`Uo6aYOZtv?Aq_S(lv{xq=Up(}jQ3Og^R1nZNWg8IA8wT_DBJH8`{*?{k@)rc!3GB#r-mgM3A$j9`8Te&IIx2laLd7Poihuc^vk;OgUD zGdO?EqPF4xAOwNkE%Wh~pU8H+$zwvJ4J1pDRp$Zg#}Q+Vk4aWY_2E~T+nYD6C!bO1 z#m6!-7N95Jw}H%|(1XWs#x=*+ygQ$~-^QqN$|wmkO{;Z*7o?3BuFqIiHx-Gj29<2m*!mAVB7Dtv-f IY^IR)e;CchWB>pF diff --git a/video-content/src/main/java/com/nuaa/video/content/config/RocketConfig.java b/video-content/src/main/java/com/nuaa/video/content/config/RocketConfig.java index 286df2d..ea86eab 100644 --- a/video-content/src/main/java/com/nuaa/video/content/config/RocketConfig.java +++ b/video-content/src/main/java/com/nuaa/video/content/config/RocketConfig.java @@ -1,11 +1,11 @@ package com.nuaa.video.content.config; import org.apache.rocketmq.client.producer.DefaultMQProducer; -import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties; import org.apache.rocketmq.spring.core.RocketMQTemplate; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.messaging.converter.MappingJackson2MessageConverter; @Configuration public class RocketConfig { @@ -22,8 +22,21 @@ public class RocketConfig { DefaultMQProducer defaultMQProducer = new DefaultMQProducer(producerGroup); defaultMQProducer.setProducerGroup(producerGroup); defaultMQProducer.setNamesrvAddr(nameServer); + // 设置消息最大长度 + defaultMQProducer.setMaxMessageSize(4 * 1024 * 1024); + + // 启用自动创建主题（注意：这需要broker端支持） + defaultMQProducer.setCreateTopicKey("AUTO_CREATE_TOPIC_KEY"); + defaultMQProducer.setSendMsgTimeout(10000); + rocketMQTemplate.setProducer(defaultMQProducer); + + // 创建并设置消息转换器 + MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter(); + converter.setSerializedPayloadClass(String.class); + converter.setStrictContentTypeMatch(false); + rocketMQTemplate.setMessageConverter(converter); + return rocketMQTemplate; } - } \ No newline at end of file diff --git a/video-content/src/main/java/com/nuaa/video/content/mq/producer/VideoMessageProducer.java b/video-content/src/main/java/com/nuaa/video/content/mq/producer/VideoMessageProducer.java index 965e8f3..2ac6af0 100644 --- a/video-content/src/main/java/com/nuaa/video/content/mq/producer/VideoMessageProducer.java +++ b/video-content/src/main/java/com/nuaa/video/content/mq/producer/VideoMessageProducer.java @@ -1,5 +1,9 @@ package com.nuaa.video.content.mq.producer; +import com.nuaa.video.content.mapper.ContentVideoMapper; +import lombok.AllArgsConstructor; +import lombok.Data; +import lombok.NoArgsConstructor; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.apache.rocketmq.client.producer.SendCallback; @@ -8,6 +12,8 @@ import org.apache.rocketmq.spring.core.RocketMQTemplate; import org.springframework.messaging.support.MessageBuilder; import org.springframework.stereotype.Component; +import java.io.Serializable; + /** * 视频消息生产者 */ @@ -17,6 +23,7 @@ import org.springframework.stereotype.Component; public class VideoMessageProducer { private final RocketMQTemplate rocketMQTemplate; + private final ContentVideoMapper videoMapper; // 定义标准化的主题名称，添加业务域前缀 private static final String TOPIC_PREFIX = "video-platform-"; @@ -89,25 +96,64 @@ public class VideoMessageProducer { public void sendVideoStatsMessage(VideoStatsMessage message) { log.info("准备发送视频统计消息: videoId={}, userId={}, action={}", message.getVideoId(), message.getUserId(), message.getAction()); + try { - SendResult sendResult = rocketMQTemplate.syncSend(VIDEO_STATS_TOPIC, - MessageBuilder.withPayload(message).build()); - log.info("视频统计消息发送成功: videoId={}, action={}, msgId={}", - message.getVideoId(), message.getAction(), sendResult.getMsgId()); + // 使用异步发送代替同步发送，避免阻塞主线程 + rocketMQTemplate.asyncSend(VIDEO_STATS_TOPIC, + MessageBuilder.withPayload(message).build(), + new SendCallback() { + @Override + public void onSuccess(SendResult sendResult) { + log.info("视频统计消息发送成功: videoId={}, action={}, msgId={}", + message.getVideoId(), message.getAction(), sendResult.getMsgId()); + } + + @Override + public void onException(Throwable throwable) { + log.error("视频统计消息发送失败: videoId={}, action={}", + message.getVideoId(), message.getAction(), throwable); + // 降级处理：直接更新数据库 + updateVideoStatsDirectly(message); + } + }); } catch (Exception e) { - log.error("视频统计消息发送失败: videoId={}, action={}", + log.error("视频统计消息发送异常，降级处理: videoId={}, action={}", message.getVideoId(), message.getAction(), e); + // 降级处理：直接更新数据库 + updateVideoStatsDirectly(message); + } + } + + private void updateVideoStatsDirectly(VideoStatsMessage message) { + try { + Long videoId = message.getVideoId(); + String action = message.getAction(); + Integer count = message.getCount() == null ? 1 : message.getCount(); + + // 直接调用数据库更新 + switch (action) { + case "play": + videoMapper.increaseViewCount(videoId, count); + log.info("直接更新数据库：视频播放计数 +{}, videoId={}", count, videoId); + break; + // 其他case处理... + } + } catch (Exception e) { + log.error("降级处理失败: 直接更新数据库出错", e); } } /** * 视频统计消息 */ - @lombok.Data - public static class VideoStatsMessage { + @NoArgsConstructor + @AllArgsConstructor + @Data + public static class VideoStatsMessage implements Serializable { + private static final long serialVersionUID = 1L; private Long videoId; private Long userId; private String action; // play, like, favorite, coin, share, danmaku - private Integer count; + private Integer count = 1; } } \ No newline at end of file diff --git a/video-content/src/main/java/com/nuaa/video/content/service/impl/VideoServiceImpl.java b/video-content/src/main/java/com/nuaa/video/content/service/impl/VideoServiceImpl.java index e2264f1..54ed2b5 100644 --- a/video-content/src/main/java/com/nuaa/video/content/service/impl/VideoServiceImpl.java +++ b/video-content/src/main/java/com/nuaa/video/content/service/impl/VideoServiceImpl.java @@ -196,18 +196,16 @@ public class VideoServiceImpl extends ServiceImpl 0); - - // 增加播放数 - videoMessageProducer.sendVideoStatsMessage( - new VideoMessageProducer.VideoStatsMessage() {{ - setVideoId(id); - setUserId(userId); - setAction("play"); - setCount(1); - }} - ); } + // 增加播放数 + VideoMessageProducer.VideoStatsMessage message = new VideoMessageProducer.VideoStatsMessage(); + message.setVideoId(id); + message.setUserId(userId); + message.setAction("play"); + message.setCount(1); + videoMessageProducer.sendVideoStatsMessage(message); + return Result.success(videoInfoVO); }