使用MediatR完成基于内存级别的消息发布订阅
在微服务架构中领域驱动模型中处理领域事件的相关操作
在区分好领域模型后，就拿代码中来说嘛，用户领域中添加用户操作可能或存在跟用户相关的一些领域事件，在添加用户的时候会执行相关的领域事件
首先需要添加nuget包 MediatR
MediatR中有几个对象：IRequest，IRequestHandler，INotificationHandler
具体用法可以看下MediatR github
领域模型需要继承Entity,里面封装了 领域事件的相关操作
public abstract class Entity
{
int? _requestedHashCode;
int _Id;
public virtual int Id
{
get
{
return _Id;
}
protected set
{
_Id = value;
}
} private List<INotification> _domainEvents;
public IReadOnlyCollection<INotification> DomainEvents => _domainEvents?.AsReadOnly(); public void AddDomainEvent(INotification eventItem)
{
_domainEvents = _domainEvents ?? new List<INotification>();
_domainEvents.Add(eventItem);
} public void RemoveDomainEvent(INotification eventItem)
{
_domainEvents?.Remove(eventItem);
} public void ClearDomainEvents()
{
_domainEvents?.Clear();
} public bool IsTransient()
{
return this.Id == default(Int32);
} public override bool Equals(object obj)
{
if (obj == null || !(obj is Entity))
return false; if (Object.ReferenceEquals(this, obj))
return true; if (this.GetType() != obj.GetType())
return false; Entity item = (Entity)obj; if (item.IsTransient() || this.IsTransient())
return false;
else
return item.Id == this.Id;
} public override int GetHashCode()
{
if (!IsTransient())
{
if (!_requestedHashCode.HasValue)
_requestedHashCode = this.Id.GetHashCode() ^ ; return _requestedHashCode.Value;
}
else
return base.GetHashCode(); }
public static bool operator ==(Entity left, Entity right)
{
if (Object.Equals(left, null))
return (Object.Equals(right, null)) ? true : false;
else
return left.Equals(right);
} public static bool operator !=(Entity left, Entity right)
{
return !(left == right);
}
}
Entity
如：添加用户的时候添加一个默认角色相关的表处理
public class TbUser : Entity, IAggregateRoot
{ public string UserName { get; set; }
public string UserPasswrod { get; set; }
public string UserPhone { get; set; }
public string XXXXX { get; set; }
[NotMapped]
public List<TbUserRole> TbUserRoles { get; set; }
public TbUser()
{
TbUserRoles = new List<TbUserRole>();
AddDomainEvent(new UserCreatedEvent() { tbUser = this });
} #region 领域职责 结合领域事件处理 黎又铭
/*
如：添加用户的时候添加默认角色 其实本身是可以直接在业务逻辑中写的
有点类是SQL触发器，添加用户的时候构建件默认角色信息模型触发事件添加角色信息 这里有注意到 实体模型并没有强制处理表接口的主外键关系 只是做了模型对象映射
*/
/// <summary>
/// 添加用户职责 只构建职责 不处理 处理交友领取事件的Command CommandHandler处理
/// </summary>
public void AddDefalutRole(TbUserRole tbUserRole)
{
TbUserRoles.Add(tbUserRole);
//添加默认角色的领域事件
AddDomainEvent(new UserRoleCreatedEvent() { TbUserRole = tbUserRole }); } #endregion }
User
先定义一个创建用户的命令操作： 这里的TbUser是领域时间模型，传递数据以及对领域模型事件操作
public class UserCreateCommand : IRequest<TbUser>
{ public TbUser tbUser{ get; set; }
}
public class UserCreateCommandHandler : IRequestHandler<UserCreateCommand, TbUser>
{ private IUserRepository _userRepository; public UserCreateCommandHandler(IUserRepository userRepository)
{
_userRepository = userRepository;
} public async Task<TbUser> Handle(UserCreateCommand request, CancellationToken cancellationToken)
{ await _userRepository.AddUserAsync(request.tbUser);
await _userRepository.UnitOfWork.SaveEntitiesAsync();
return request.tbUser;
}
}
这里还需要一个MediatR的扩展，发布领域事件
static class MediatorExtension
{
/// <summary>
/// 异步处理领域事件
/// </summary>
/// <param name="mediator"></param>
/// <param name="ctx"></param>
/// <returns></returns>
public static async Task DispatchDomainEventsAsync(this IMediator mediator, UserDbContext ctx)
{
var domainEntities = ctx.ChangeTracker
.Entries<Entity>()
.Where(x => x.Entity.DomainEvents != null && x.Entity.DomainEvents.Any()); var domainEvents = domainEntities
.SelectMany(x => x.Entity.DomainEvents)
.ToList(); domainEntities.ToList()
.ForEach(entity => entity.Entity.ClearDomainEvents()); var tasks = domainEvents
.Select(async (domainEvent) => {
await mediator.Publish(domainEvent);
}); await Task.WhenAll(tasks);
}
}
扩展
下面是订阅领域事件
public class UserRoleNotification : INotificationHandler<UserRoleCreatedEvent>
{
private IUserRepository _userRepository;
public UserRoleNotification(IUserRepository userRepository)
{
_userRepository = userRepository;
}
public async Task Handle(UserRoleCreatedEvent notification, CancellationToken cancellationToken)
{ await _userRepository.AddUserRoleAsync(notification.TbUserRole);
}
在controller中的使用：
[HttpPost]
[Route("adduser")]
public async Task<IActionResult> AddUser([FromBody] TbUser tbUser)
{
tbUser.AddDefalutRole(new TbUserRole { RoleId = , UserId = });
var command = new UserCreateCommand() { tbUser = tbUser };
var result = await _mediator.Send(command);
return Ok(result);
}
看下效果